Python七牛异步SDK-async_cow
Async Cow Python 七牛异步SDK
python七牛异步sdk,gitee:https://gitee.com/xixigroup/async_cow
本SDK基于官方SDK改造而成,但又对其进行了进一步封装,简化了相关操作 例如:
1、不需要使用者关心token问题
2、简化了相关导包和引用,并且保持接口一致
3、实现了异步IO,引入协程概念,IO层面引入aiohttp,aiofiles等,使得本SDK适用于异步编程
感谢您的星星?
[官方SDK请见] https://developer.qiniu.com/kodo/1242/python
组织:
QQ群: 614714752
Install
python解释器版本要求:> 3.6
# 标准安装 pip install async_cow # 从官方源安装,你能获取最新版本SDK pip install async_cow -i https://pypi.python.org/simple
Usage
初始化
在你需要的地方
from async_cow import AsyncCow, ClientCow cow = AsyncCow(, ) client = ClientCow( , )
云存储桶操作
b = cow.get_bucket()
后面都用这个桶对象来操作。 它代表了
列出所有的bucket
res = await b.buckets()
列出一个bucket中的所有文件
res = await b.list()
这个方法还有 marker, limit, prefix这三个可选参数,详情参考官方文档
bucket相关方法和用法和官方SDK同步
上传
file_path = '/Users/admin/Desktop/123.jpg'
with open(file_path, 'rb') as f:
c = f.read()
# 上传二进制流
res = await b.put_data(
key='AK47.jpg', # 上传后的文件名
data=c
)
# 上传文件
res = await b.put_file(
key='AK472.jpg', # 上传后的文件名
file_path=file_path
)
for i in res:
print(i)
删除,查看文件信息
await b.stat('a') # 查看单个文件信息
await b.delete('a') # 删除单个文件
拷贝,移动(改名)经测试,只能在桶内copy和move
这两个操作需要提供源文件名和目标文件名
await b.copy('a', 'b') # 将'a' 拷贝至'b'
await b.move('a', 'b') # 将'a' 改名为'b'
有没有觉得比官方SDK容易使用多呢?
异常
在封装aiohttp操作时已经处理了大部分异常和重试过程,但仍会存在一些意外
所以安全的做法是这样:
try:
await b.delete('a')
except Exception as e:
# 自行处理
pass
短信客户端操作
# 获取一个短信客户端对象 sms = cow.get_sms()
创建,查看,编辑,删除签名
await sms.createSignature(, ) await sms.querySignature() await sms.updateSignature( , ) await sms.deleteSignature( )
创建,查看,编辑,删除模版
await sms.createTemplate(, , , , ) await sms.queryTemplate( ) await sms.updateTemplate( , , , , ) await sms.deleteTemplate( )
发送短信,查看发送记录,查询发送计费条数
await sms.sendMessage(, , ) await sms.get_messages_info() await sms.get_charge_message_count( , , , )
持久化
# 获取持久化类对象 p = cow.get_persistent_fop()
文件持久化
await p.execute(, )
直播连麦管理
# 获取一个管理类对象 r = cow.get_rtc_server()
创建,获取,修改,删除app
await r.create_app() await r.get_app() await r.update_app(, ) await r.delete_app( )
用户列表,踢出用户
await r.list_user(, ) await r.kick_user( , , )
查看活跃房间
await r.list_active_rooms()
CDN管理
# 获取cdn管理类对象 cdn = cow.get_cdn_manager()
刷新文件、目录
await cdn.refresh_urls() await cdn.refresh_dirs( ) # 同时刷新urls和dirs await cdn.refresh_urls_and_dirs( , )
预取文件列表
await cdn.prefetch_urls()
查询宽带、流量数据
# 宽带 await cdn.get_bandwidth_data(, , , ) # 流量 await cdn.get_flux_data( , , , )
获取日志下载链接
await cdn.get_log_list_data(, )
修改证书
await cdn.put_httpsconf(, )
域名管理
# 获取域名管理类对象 d = cow.get_domain_manager()
创建,查看,删除域名
await d.create_domain(, ) await d.get_domain( ) await d.delete_domain( )
上线、下线域名
await d.domain_online() await d.domain_offline( )
创建、修改证书
await d.create_sslcert(, , ) await d.put_httpsconf( , , , )
账号客户端
# 获取帐号客户端对象 app = client.get_app()
创建、获取管理客户端
await app.create_qcos_client() await app.get_qcos_client( )
账号密钥
# 获取帐号下应用的密钥 await app.get_app_keys() # 获取帐号下可用的应用的密钥 await app.get_valid_app_auth( )
当前账号的信息
await app.get_account_info()
获得指定应用所在区域的产品信息
await app.get_app_region_products()
获取指定区域产品信息
await app.get_region_products()
获得账号可见的区域的信息
await app.list_regions()
创建、获得、删除当前账号的应用
await app.create_app() await app.list_apps() await app.delete_app( )
资源管理客户端
# 获取资源管理客户端对象 q = client.get_qcos_client()
创建、获取、删除服务组
await q.create_stack() await q.get_stack( ) await q.delete_stack( )
启动、停止服务组
await q.start_stack() await q.stop_stack( )
创建、获取、更新删除服务
await q.create_service(, ) # 查看服务 await q.get_service_inspect( , ) # 获得服务列表 await q.list_services( ) await q.update_service( , , ) await q.delete_service( , )
启动、停止服务
await q.start_service(, ) await q.stop_service( , )
扩容、缩容服务
await q.scale_service(, , )
创建、删除、扩容存储卷
await q.create_service_volume(, , ) await q.delete_service_volume( , , ) await q.extend_service_volume( , , , )
查看、列出容器
await q.get_container_inspect() await q.list_containers()
启动、停止、重启容器
await q.start_container() await q.stop_container( ) await q.restart_container( )
接入点
# 列出接入点 await q.list_aps() # 搜索接入点 await q.search_ap(, ) # 查看接入点 await q.get_ap( ) # 申请接入点 await q.create_ap( ) # 更新接入点 await q.update_ap( , ) # 更新接入点端口配置 await q.set_ap_port( , , ) # 释放接入点 await q.delete_ap( )
自定义域名
# 绑定自定义域名 await q.publish_ap(, ) # 解绑自定义域名 await q.unpublish_ap( , )
查看健康检查结果
await q.get_ap_port_healthcheck(, )
调整后端实例配置
await q.set_ap_port_container(, , )
接入点端口
# 临时关闭接入点端口 await q.disable_ap_port(, ) # 开启接入点端口 await q.enable_ap_port( , )
列出入口提供商
await q.get_ap_providers()
获取一次性代理地址
await q.get_web_proxy()
测试
桶测试
- 首先从github clone项目到本地
- 在项目中有example供测试,配置好main.py中的access_key、secret_key、bucket以及file_path参数后即可开始测试
特别鸣谢:Hagworm 以及七牛官方
欢迎大佬指正!感谢您的星星?