wechatpy.client.api.cloud 源代码

# -*- coding: utf-8 -*-

import json
from enum import IntEnum

import requests

from wechatpy.client.api.base import BaseWeChatAPI


class FileType(IntEnum):
    JSON = 1
    CSV = 2


class ConflictMode(IntEnum):
    INSERT = 1
    UPSERT = 2


[文档]class WeChatCloud(BaseWeChatAPI): API_BASE_URL = "https://api.weixin.qq.com/"
[文档] def invoke_cloud_function(self, env, name, data=None): """ 触发云函数 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/functions/invokeCloudFunction.html :param env: 云开发环境 ID :param name: 云函数名称 :param data: 云函数的传入参数,具体结构由开发者定义 """ return self._post( "tcb/invokecloudfunction", params={"env": env, "name": name,}, data=data, result_processor=lambda x: json.loads(x["resp_data"]), )
[文档] def db_import(self, env, collection, file_path, file_type, conflict_mode, stop_on_error=True): """ 数据库导入 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseMigrateImport.html :param env: 云开发环境 ID :param collection: 导入 collection 名称 :param file_path: 导入文件路径(导入文件需先上传到同环境的存储中,可使用开发者工具或 HTTP API的上传文件 API上传) :param file_type: 导入文件类型,文件格式参考数据库导入指引中的文件格式部分,值为数字,1 为 JSON,2 为 CSV :param stop_on_error: 是否在遇到错误时停止导入,默认为 True :param conflict_mode: 冲突处理模式, 值为数字,1 为 INSERT,2 为 UPSERT :return: 导入任务 ID,可使用数据库迁移进度查询 API 查询导入进度及结果 """ return self._post( "tcb/databasemigrateimport", data={ "env": env, "collection_name": collection, "file_path": file_path, "file_type": file_type, "stop_on_error": stop_on_error, "conflict_mode": conflict_mode, }, result_processor=lambda x: x["job_id"], )
[文档] def db_export(self, env, file_path, file_type, query): """ 数据库导出 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseMigrateExport.html :param env: 云开发环境 ID :param file_path: 导出文件路径(导入文件需先上传到同环境的存储中,可使用开发者工具或 HTTP API的上传文件 API上传) :param file_type: 导出文件类型,文件格式参考数据库导入指引中的文件格式部分,值为数字,1 为 JSON,2 为 CSV :param query: 导出条件 :return: 导出任务 ID,可使用数据库迁移进度查询 API 查询导出结果,获取文件下载链接 """ return self._post( "tcb/databasemigrateexport", data={"env": env, "file_path": file_path, "file_type": file_type, "query": query,}, result_processor=lambda x: x["job_id"], )
[文档] def db_query_migrate_info(self, env, job_id): """ 数据库迁移状态查询 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseMigrateQueryInfo.html :param env: 云开发环境 ID :param job_id: 任务 ID """ return self._post("tcb/databasemigratequeryinfo", data={"env": env, "job_id": job_id,})
[文档] def db_update_index(self, env, collection, create_indexes=None, drop_indexes=None): """ 变更数据库索引 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/updateIndex.html :param env: 云开发环境 ID :param collection: 导出 collection 名称 """ assert create_indexes or drop_indexes return self._post( "tcb/updateindex", data={ "env": env, "collection_name": collection, "create_indexes": create_indexes or [], "drop_indexes": create_indexes or [], }, )
[文档] def db_add_collection(self, env, collection): """ 新增集合 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseCollectionAdd.html :param env: 云开发环境 ID :param collection: 集合名称 """ return self._post("tcb/databasecollectionadd", data={"env": env, "collection_name": collection,},)
[文档] def db_delete_collection(self, env, collection): """ 删除集合 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseCollectionDelete.html :param env: 云开发环境 ID :param collection: 集合名称 """ return self._post("tcb/databasecollectiondelete", data={"env": env, "collection_name": collection,},)
[文档] def db_get_collection(self, env, offset=0, limit=10): """ 获取特定云环境下集合信息 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseCollectionGet.html :param env: 云开发环境 ID :param offset: 偏移量,默认为 0 :param limit: 获取数量限制, 默认为 10 """ return self._post("tcb/databasecollectionget", data={"env": env, "offset": offset, "limit": limit,},)
[文档] def db_add(self, env, query): """ 数据库插入记录 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseAdd.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回插入成功的数据集合主键 _id 列表 """ return self._post( "tcb/databaseadd", data={"env": env, "query": query,}, result_processor=lambda x: x["id_list"], )
[文档] def db_delete(self, env, query): """ 数据库删除记录 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseDelete.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回删除记录数量 """ return self._post( "tcb/databasedelete", data={"env": env, "query": query,}, result_processor=lambda x: x["deleted"], )
[文档] def db_update(self, env, query): """ 数据更新记录 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseUpdate.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回的 JSON 数据包 """ return self._post("tcb/databaseupdate", data={"env": env, "query": query,})
[文档] def db_query(self, env, query): """ 数据库查询记录 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseQuery.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回的 JSON 数据包 """ return self._post("tcb/databasequery", data={"env": env, "query": query,})
[文档] def db_aggregate(self, env, query): """ 数据库聚合 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseAggregate.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回记录列表 """ return self._post( "tcb/databaseaggregate", data={"env": env, "query": query,}, result_processor=lambda x: x["data"], )
[文档] def db_count(self, env, query): """ 统计集合记录数或统计查询语句对应的结果记录数 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/database/databaseCount.html :param env: 云开发环境 ID :param query: 数据库操作语句 :return: 返回记录数量 """ return self._post( "tcb/databasecount", data={"env": env, "query": query,}, result_processor=lambda x: x["count"], )
[文档] def upload_file(self, env, path): """ 获取文件上传链接 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/storage/uploadFile.html :param env: 云开发环境 ID """ with open(path, "rb") as f: res = self._post("tcb/uploadfile", data={"env": env, "path": path,}) signature = res["authorization"] token = res["token"] cos_file_id = res["cos_file_id"] upload_res = requests.post( res["url"], files={ "key": path, "Signature": signature, "x-cos-security-token": token, "x-cos-meta-fileid": cos_file_id, # 注意!file 字段须放在最后,否则上传大文件会失败 "file": f, }, ) upload_res.raise_for_status() return upload_res
[文档] def download_files(self, env, file_list): """ 获取文件下载链接 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/storage/batchDownloadFile.html :param env: 云开发环境 ID :param file_list: 文件列表 :return: 返回文件列表 """ return self._post( "tcb/batchdownloadfile", data={"env": env, "file_list": file_list,}, result_processor=lambda x: x["file_list"], )
[文档] def delete_files(self, env, fileid_list): """ 删除文件 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/storage/batchDeleteFile.html :param env: 云开发环境 ID :param fileid_list: 文件 ID 列表 :return: 被删除的文件列表 """ return self._post( "tcb/batchdeletefile", data={"env": env, "fileid_list": fileid_list,}, result_processor=lambda x: x["delete_list"], )
[文档] def get_qcloud_token(self, lifespan=7200): """ 获取腾讯云 API 调用凭证 详情请参考 https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/utils/getQcloudToken.html """ return self._post("tcb/getqcloudtoken", data={"lifespan": lifespan})