企业微信或钉钉的access_token缓存方案
需求背景
https://developer.work.weixin.qq.com/document/path/91039
调用企业微信API或钉钉API首先需要获取access_token，相当于创建了一个登录凭证，其它的业务API接口，都需要依赖于access_token来鉴权调用者身份。
开发者需要缓存access_token，用于后续接口的调用（注意：不能频繁调用gettoken接口，否则会受到频率拦截）。当access_token失效或过期时，需要重新获取。
企业微信
缓存文件方案
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
| class WeChat:
def __init__(self):
self.CORPID = ""
self.CORPSECRET = ""
self.AGENTID = ""
def _get_access_token(self):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
payload = {'corpid': self.CORPID,
'corpsecret': self.CORPSECRET,
}
req = requests.post(url, params=payload)
return req.json()["access_token"]
def get_access_token(self):
"""
当已缓存token的话，直接返回token
当获取失败，去请求接口获取token
文件缓存在文件中
"""
try:
with open('access_token.conf', 'r') as f:
t, access_token = f.read().split()
except:
with open('access_token.conf', 'w') as f:
access_token = self._get_access_token()
cur_time = time.time()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
else:
cur_time = time.time()
if 0 < cur_time - float(t) < 7260:
return access_token
else:
with open('access_token.conf', 'w') as f:
access_token = self._get_access_token()
f.write('\t'.join([str(cur_time), access_token]))
return access_token
|
缓存到redis
redis_tool.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| import redis
from config.config import OpsConfig as Config
class RedisTool(object):
def __init__(self):
pass
@classmethod
def get_redis_client(cls):
redis_pool = redis.ConnectionPool(host=Config.REDIS_HOST,
port=Config.REDIS_PORT,
password=Config.REDIS_PASSWORD,
db=Config.REDIS_DB)
redis_client = redis.Redis(connection_pool=redis_pool)
return redis_client
|
qiyeweixin.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
| import requests
from common.redis_tool import RedisTool
class WeChat:
def __init__(self):
self.CORPID = ""
self.CORPSECRET = ""
self.AGENTID = ""
def _get_access_token(self):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
payload = {'corpid': self.CORPID,
'corpsecret': self.CORPSECRET,
}
req = requests.post(url, params=payload)
return req.json()["access_token"]
def get_access_token(self):
"""
当已缓存token的话，直接返回token
当获取失败，去请求接口获取token
"""
redis = RedisTool()
redis_client = redis.get_redis_client()
key = 'wechat_token'
token = redis_client.get(key)
if token:
return token
else:
token = self._get_access_token()
redis_client.set(key, token, 7200)
return token
|