Async Cow Python 七牛異步SDK
本SDK基于官方SDK改造而成，但又對(duì)其進(jìn)行了進(jìn)一步封裝，簡(jiǎn)化了相關(guān)操作
例如：
1、不需要使用者關(guān)心token問題
2、簡(jiǎn)化了相關(guān)導(dǎo)包和引用，并且保持接口一致
3、實(shí)現(xiàn)了異步IO，引入?yún)f(xié)程概念，IO層面引入aiohttp，aiofiles等，使得本SDK適用于異步編程
gitee:https://gitee.com/xixigroup/async_cow
感謝您的星星?
[官方SDK請(qǐng)見] https://developer.qiniu.com/kodo/1242/python
組織:
QQ群: 614714752
Install
python解釋器版本要求：> 3.6
# 標(biāo)準(zhǔn)安裝
pip install async_cow
# 從官方源安裝，你能獲取最新版本SDK
pip install async_cow -i https://pypi.python.org/simple
Usage
初始化
在你需要的地方
from async_cow import AsyncCow, ClientCow
cow = AsyncCow(<ACCESS_KEY>, <SECRET_KEY>)
client = ClientCow(<ACCESS_KEY>, <SECRET_KEY>)
云存儲(chǔ)桶操作
b = cow.get_bucket(<BUCKET>)
后面都用這個(gè)桶對(duì)象來操作。 它代表了<BUCKET>
列出所有的bucket
res = await b.buckets()
列出一個(gè)bucket中的所有文件
res = await b.list()
這個(gè)方法還有 marker, limit, prefix這三個(gè)可選參數(shù)，詳情參考官方文檔
bucket相關(guān)方法和用法和官方SDK同步
上傳
file_path = '/Users/admin/Desktop/123.jpg'
with open(file_path, 'rb') as f:
c = f.read()
# 上傳二進(jìn)制流
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') # 查看單個(gè)文件信息
await b.delete('a') # 刪除單個(gè)文件
拷貝，移動(dòng)（改名）經(jīng)測(cè)試，只能在桶內(nèi)copy和move
這兩個(gè)操作需要提供源文件名和目標(biāo)文件名
await b.copy('a', 'b') # 將'a' 拷貝至'b'
await b.move('a', 'b') # 將'a' 改名為'b'
有沒有覺得比官方SDK容易使用多呢？
異常
在封裝aiohttp操作時(shí)已經(jīng)處理了大部分異常和重試過程，但仍會(huì)存在一些意外
所以安全的做法是這樣：
try:
await b.delete('a')
except Exception as e:
# 自行處理
pass
短信客戶端操作
# 獲取一個(gè)短信客戶端對(duì)象
sms = cow.get_sms()
創(chuàng)建，查看，編輯，刪除簽名
await sms.createSignature(<signature>, <source>)
await sms.querySignature()
await sms.updateSignature(<id>, <signature>)
await sms.deleteSignature(<id>)
創(chuàng)建，查看，編輯，刪除模版
await sms.createTemplate(<name>, <template>, <type>, <description>, <signature_id>)
await sms.queryTemplate(<audit_status>)
await sms.updateTemplate(<id>, <name>, <template>, <description>, <signature_id>)
await sms.deleteTemplate(<id>)
發(fā)送短信，查看發(fā)送記錄，查詢發(fā)送計(jì)費(fèi)條數(shù)
await sms.sendMessage(<template_id>, <mobiles>, <parameters>)
await sms.get_messages_info()
await sms.get_charge_message_count(<start>, <end>, <g>, <status>)
持久化
# 獲取持久化類對(duì)象
p = cow.get_persistent_fop(<bucket>)
文件持久化
await p.execute(<key>, <fops>)
直播連麥管理
# 獲取一個(gè)管理類對(duì)象
r = cow.get_rtc_server()
創(chuàng)建，獲取，修改，刪除app
await r.create_app(<data>)
await r.get_app()
await r.update_app(<app_id>, <data>)
await r.delete_app(<app_id>)
用戶列表，踢出用戶
await r.list_user(<app_id>, <data>)
await r.kick_user(<app_id>, <data>, <user_id>)
查看活躍房間
await r.list_active_rooms(<app_id>)
CDN管理
# 獲取cdn管理類對(duì)象
cdn = cow.get_cdn_manager()
刷新文件、目錄
await cdn.refresh_urls(<urls>)
await cdn.refresh_dirs(<dirs>)
# 同時(shí)刷新urls和dirs
await cdn.refresh_urls_and_dirs(<urls>, <dirs>)
預(yù)取文件列表
await cdn.prefetch_urls(<urls>)
查詢寬帶、流量數(shù)據(jù)
# 寬帶
await cdn.get_bandwidth_data(<domains>, <start_date>, <end_date>, <granularity>)
# 流量
await cdn.get_flux_data(<domains>, <start_date>, <end_date>, <granularity>)
獲取日志下載鏈接
await cdn.get_log_list_data(<domains>, <log_date>)
修改證書
await cdn.put_httpsconf(<name>, <certid>)
域名管理
# 獲取域名管理類對(duì)象
d = cow.get_domain_manager()
創(chuàng)建，查看，刪除域名
await d.create_domain(<name>, <body>)
await d.get_domain(<name>)
await d.delete_domain(<name>)
上線、下線域名
await d.domain_online(<name>)
await d.domain_offline(<name>)
創(chuàng)建、修改證書
await d.create_sslcert(<name>, <certid>, <forceHttps>)
await d.put_httpsconf(<name>, <common_name>, <pri>, <ca>)
賬號(hào)客戶端
# 獲取帳號(hào)客戶端對(duì)象
app = client.get_app()
創(chuàng)建、獲取管理客戶端
await app.create_qcos_client(<app_uri>)
await app.get_qcos_client(<app_uri>)
賬號(hào)密鑰
# 獲取帳號(hào)下應(yīng)用的密鑰
await app.get_app_keys(<app_uri>)
# 獲取帳號(hào)下可用的應(yīng)用的密鑰
await app.get_valid_app_auth(<app_uri>)
當(dāng)前賬號(hào)的信息
await app.get_account_info()
獲得指定應(yīng)用所在區(qū)域的產(chǎn)品信息
await app.get_app_region_products(<app_uri>)
獲取指定區(qū)域產(chǎn)品信息
await app.get_region_products(<region>)
獲得賬號(hào)可見的區(qū)域的信息
await app.list_regions()
創(chuàng)建、獲得、刪除當(dāng)前賬號(hào)的應(yīng)用
await app.create_app(<args>)
await app.list_apps()
await app.delete_app(<app_uri>)
資源管理客戶端
# 獲取資源管理客戶端對(duì)象
q = client.get_qcos_client()
創(chuàng)建、獲取、刪除服務(wù)組
await q.create_stack(<args>)
await q.get_stack(<stack>)
await q.delete_stack(<stack>)
啟動(dòng)、停止服務(wù)組
await q.start_stack(<stack>)
await q.stop_stack(<stack>)
創(chuàng)建、獲取、更新刪除服務(wù)
await q.create_service(<stack>, <args>)
# 查看服務(wù)
await q.get_service_inspect(<stack>, <service>)
# 獲得服務(wù)列表
await q.list_services(<stack>)
await q.update_service(<stack>, <service>, <args>)
await q.delete_service(<stack>, <service>)
啟動(dòng)、停止服務(wù)
await q.start_service(<stack>, <service>)
await q.stop_service(<stack>, <service>)
擴(kuò)容、縮容服務(wù)
await q.scale_service(<stack>, <service>, <args>)
創(chuàng)建、刪除、擴(kuò)容存儲(chǔ)卷
await q.create_service_volume(<stack>, <service>, <args>)
await q.delete_service_volume(<stack>, <service>, <volume>)
await q.extend_service_volume(<stack>, <service>, <volume>, <args>)
查看、列出容器
await q.get_container_inspect(<ip>)
await q.list_containers()
啟動(dòng)、停止、重啟容器
await q.start_container(<ip>)
await q.stop_container(<ip>)
await q.restart_container(<ip>)
接入點(diǎn)
# 列出接入點(diǎn)
await q.list_aps()
# 搜索接入點(diǎn)
await q.search_ap(<mode>, <query>)
# 查看接入點(diǎn)
await q.get_ap(<apid>)
# 申請(qǐng)接入點(diǎn)
await q.create_ap(<args>)
# 更新接入點(diǎn)
await q.update_ap(<apid>, <args>)
# 更新接入點(diǎn)端口配置
await q.set_ap_port(<apid>, <port>, <args>)
# 釋放接入點(diǎn)
await q.delete_ap(<apid>)
自定義域名
# 綁定自定義域名
await q.publish_ap(<apid>, <args>)
# 解綁自定義域名
await q.unpublish_ap(<apid>, <args>)
查看健康檢查結(jié)果
await q.get_ap_port_healthcheck(<apid>, <port>)
調(diào)整后端實(shí)例配置
await q.set_ap_port_container(<apid>, <port>, <args>)
接入點(diǎn)端口
# 臨時(shí)關(guān)閉接入點(diǎn)端口
await q.disable_ap_port(<apid>, <port>)
# 開啟接入點(diǎn)端口
await q.enable_ap_port(<apid>, <port>)
列出入口提供商
await q.get_ap_providers()
獲取一次性代理地址
await q.get_web_proxy(<backend>)
測(cè)試
桶測(cè)試
- 首先從github clone項(xiàng)目到本地
- 在項(xiàng)目中有example供測(cè)試，配置好main.py中的access_key、secret_key、bucket以及file_path參數(shù)后即可開始測(cè)試
特別鳴謝：Hagworm 以及七牛官方
歡迎大佬指正！感謝您的星星?
本文由【希希大隊(duì)長(zhǎng)】原創(chuàng)
