Python实现企业微信Token自动获取到SQLite存储

背景

我们有很多跟企业微信对接数据集成的项目，企微的access_token是有有效期的，一个应用的token可能会多个任务共用，实现了一个工具类来保存access_token

环境准备

• Python 3.6+
• 所需库：
  • requests（用于HTTP请求）
  • sqlite3（Python内置数据库操作）

3. 实现方案

数据库设计

我们使用SQLite作为Token存储方案，具有以下优点：

• 单文件数据库，无需额外服务
• 轻量级，适合小型应用
• 支持基本的SQL操作

直接上代码

#-*- coding:utf-8 -*-
#Author：Evan
#Mail：evan#chengwenit.com

#管理企微应用token

import requests
import cjcconfig
import time
import chw_sqllite_client

class workwechatapp():
    # 从企业微信获取token写入sqllite数据库的app表
    def get_qwtoken_cjcdb(self,corpid,appsecret,appname):
        """
        获取企业微信应用access_token
        appsecret：应用密钥
        appid：应用id
        return：access_token
        """
        params = {'corpId': corpid,
                'corpSecret': appsecret}
        response = requests.get(url=cjcconfig.get_token_url,params=params)
        if response.status_code == 200:
            response = response.json()
            if response['errcode'] == 0:
                print('获取token成功')
                sql = chw_sqllite_client.sqllitToolSql(cjcconfig.sqldbname)
                sql.execute("REPLACE INTO chw_qw_app (appsecret,apptoken,synctime,memo) values(?,?,?,?);",(appsecret,response['access_token'],str(time.time()),appname))
                if sql.execute:
                    print("更新token成功")
                    sql.close()
                    return True
                else:
                    print("更新token失败")
                    sql.close()
                    return False
            else:
                print('获取token失败')
                return False
        else:
            print('请求失败')
            return False   

    # 根据appname从sqllite数据库企微APP表中获取token
    def get_cjc_qw_app_token(self,appname):
        # 获取企业微信token
        sql = chw_sqllite_client.sqllitToolSql(cjcconfig.sqldbname)
        res = sql.query("select apptoken from chw_qw_app where memo=?;",(appname,))
        if res is None:
            return False
        else:
            return res[0][0]

if __name__ == "__main__":
    """
    测试代码
    """

    # 从企业微信获取token更新到sqllite中
    #x = workwechatapp()
    #applist = cjcconfig.applist
    #for i in range(len(applist)):
    #    x.get_qwtoken_cjcdb(cjcconfig.corpId,applist[i]["appsecret"],applist[i]["appname"])
    ## 从sqllite中获取token
    #a = x.get_cjc_qw_app_token("企微通讯录")
    #b = x.get_cjc_qw_app_token("企微成集云")
    #print(a)
    #print(b)
    ```
## 注意事项
- Token过期处理
- 数据库文件权限
- 并发访问考虑
- 错误处理与重试机制