代码收藏家技术教程 2022-08-04

python操作小程序云开发数据库进行增删改查

文章目录

1、获取access_token

2、通过数据库api进行数据库操作

3、运行结果

最近想把本地数据库中的部分数据同步到小程序的云数据库中。由于服务端为python，因此想通过python同步读取后写入到小程序的云数据库中会更加方便。

参看了小程序的云开发中提供了一系列的
HTTP api 相关文档，很快就能实现对云开发数据库的增删改查了。大致分如下两步走。

1、获取access_token

调用云数据库必要的两个参数是小程序的appid和appsecret。这两个值可以登录自己的小程序后台去获取。access_token两小时有效，超时则重新请求获取,因此这里我建了一个json文件保存最近的token和更新时间。

def get_access_token():
    """"
    获取access_token
    """
    with open("token.json", "r") as f:
        config = json.loads(f.read())
    now_time = int(time.time())
    # access_token两小时有效，超时则重新请求获取
    if now_time - config.get("update_time", 0) >= 7200:
        appid = '******'  # 小程序ID
        appsecret = '******'  # 小程序秘钥
		wechart_url = f'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={appid}&secret={appsecret}'
        response = requests.get(wechart_url)
        result = response.json()
        config = {
            "token": result["access_token"],
            "update_time": now_time
        }
        with open("token.json", "w") as f:
            f.write(json.dumps(config))

    return config["token"]  # 将返回值解析获取access_token

2、通过数据库api进行数据库操作

获取了access_token后就能进行增删改成数据库集合和记录的一系列操作了。
每次记录查询需要的查询为小程序的环境id env和query查询语句。增删改查对应不同的url，均为post请求。构造参数查询即可。

class Cloud:
    def __init__(self, env, collection_name):
        self.access_token = get_access_token()
        self.query_url = 'https://api.weixin.qq.com/tcb/databasequery?access_token={}'.format(self.access_token)
        self.add_url = 'https://api.weixin.qq.com/tcb/databaseadd?access_token={}'.format(self.access_token)
        self.update_url = 'https://api.weixin.qq.com/tcb/databaseupdate?access_token={}'.format(self.access_token)
        self.delete_url = 'https://api.weixin.qq.com/tcb/databasedelete?access_token={}'.format(self.access_token)
        self.env = env
        self.collection_name = collection_name
        self.post_data = {"env": self.env}  # 请求参数，每次的请求参数需要env环境id和query查询语句

    def query(self, search_param):
        """
        search_param: dict
        """
        self.post_data["query"] = f"db.collection('{self.collection_name}').where({search_param}).get()"
        res = requests.post(self.query_url, data=json.dumps(self.post_data))
        return res.json()

    def update(self, search_param, update_dict):
        """
        search_param:dict 查询要更新的某条记录
        update_dict:dict 要修改的值
        """
        update_data = "{data:%s}" % update_dict
        self.post_data["query"] = f"db.collection('{self.collection_name}').where({search_param}).update({update_data})"
        response = requests.post(self.update_url, data=json.dumps(self.post_data))
        result = response.json()
        return result

    def add(self, new_data):
        """
        new_data: list of dict
        """
        new_data = "{data:%s}" % new_data
        self.post_data["query"] = f"db.collection('{self.collection_name}').add({new_data})"
        response = requests.post(self.add_url, data=json.dumps(self.post_data))
        result = response.json()
        # 执行成功返回状态码0
        if result["errcode"] == 0:
            return result['id_list']

    def delete(self, search_param):
        self.post_data["query"] = f"db.collection('{self.collection_name}').where({search_param}).remove()"
        res = requests.post(self.delete_url, data=json.dumps(self.post_data))
        return res.json()

3、运行结果

接着写上测试增删改查的主函数

if __name__ == '__main__':
    _env = "*****"# 自己的云开发环境id
    _collection_name = "projects"# 需要查询的数据集合名称
    # 新增
    Cloud(_env, _collection_name).add([{
        "name": "test",
        "price": 0,
        "language": "python"
    }])
    # 修改
    print(Cloud(_env, _collection_name).update({"name": "test"}, {"name": "update_test"}))
    # 修改后查询
    print(Cloud(_env, _collection_name).query({"name": "update_test"}))
    # 删除
    print(Cloud(_env, _collection_name).delete({"name": "update_test"}))
    # 删除后查询
    print(Cloud(_env, _collection_name).query({}))

运行结果如下，整体流程搞定。