Python SDK
从 v5.0.0 版本开始,我们对 SDK 的内容进行了精简。所有管理操作,比如:创建/删除 bucket、为 bucket 绑定域名、设置数据处理的样式分隔符、新增数据处理样式等都去除了,统一建议到七牛开发者平台来完成。另外,此前服务端还有自己独有的上传 API,现在也推荐统一成基于客户端上传的工作方式。
此 Python SDK 适用于 2.7 和 3.3及以上版本
安装
- 直接安装:
pip install qiniu
或
easy_install qiniu
- 源码安装:
#从Python SDK 下载地址下载源码
tar xvzf python-sdk-$VERSION.tar.gz
cd python-sdk-$VERSION
python setup.py install
初始化
在使用SDK 前,您需要一对有效的 AccessKey 和 SecretKey 签名授权。
可以通过如下步骤获得:
获取Access Key 和 Secret Key 后,调用如下两行代码进行初始化对接:
from qiniu import Auth
q = Auth(access_key, secret_key)
上传相关
上传流程
为了尽可能地改善终端用户的上传体验,七牛云存储首创了客户端直传功能。更多信息请参阅业务流程。
SDK 内置两种上传方式:表单上传和分片上传,并根据具体情况,内部做了自动切换。表单上传使用一个 HTTP POST 请求完成文件的上传,因此比较适合较小的文件和较好的网络环境。相比而言,分片上传更能适应不稳定的网络环境,也比较适合上传比较大的文件(例如数百 MB 或更大)。
若需深入了解上传方式之间的区别,请参阅上传类型中 表单上传 、分片上传 v1 版 和 分片上传 v2 版 接口说明。 可以选择分片上传版本,推荐 version = 'v2'
,表示 分片上传 v2 版,默认 version = 'v1'
,表示 分片上传 v1 版,兼容历史情况,详见 put_file 或 put_stream 方法的 API 文档 。
上传代码:
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth, put_file, etag
import qiniu.config
#需要填写你的 Access Key 和 Secret Key
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#构建鉴权对象
q = Auth(access_key, secret_key)
#要上传的空间
bucket_name = 'Bucket_Name'
#上传后保存的文件名
key = 'my-python-logo.png'
#生成上传 Token,可以指定过期时间等
token = q.upload_token(bucket_name, key, 3600)
#要上传文件的本地路径
localfile = './sync/bbb.jpg'
ret, info = put_file(token, key, localfile, version='v2')
print(info)
assert ret['key'] == key
assert ret['hash'] == etag(localfile)
上传&回调
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth, put_file, etag
import qiniu.config
access_key = 'Access_Key'
secret_key = 'Secret_Key'
q = Auth(access_key, secret_key)
bucket_name = 'Bucket_Name'
key = 'my-python-logo.png'
#上传文件到存储后, 存储服务将文件名和文件大小回调给业务服务器。
policy={
'callbackUrl':'http://your.domain.com/callback.php',
'callbackBody':'filename=$(fname)&filesize=$(fsize)'
}
token = q.upload_token(bucket_name, key, 3600, policy)
localfile = './sync/bbb.jpg'
ret, info = put_file(token, key, localfile, version='v2')
print(info)
assert ret['key'] == key
assert ret['hash'] == etag(localfile)
上传&预转持续化
开发者希望在上传文件过程中自动触发数据处理过程(即预转持久化)。
以视频转码为例:
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth, put_file, etag, urlsafe_base64_encode
import qiniu.config
access_key = 'Access_Key'
secret_key = 'Secret_Key'
q = Auth(access_key, secret_key)
bucket_name = 'Bucket_Name'
key = 'my-python-logo.png'
#设置转码参数
fops = 'avthumb/mp4/s/640x360/vb/1.25m'
#转码是使用的队列名称
pipeline = 'abc'
#可以对转码后的文件进行使用saveas参数自定义命名,当然也可以不指定文件会默认命名并保存在当前空间
saveas_key = urlsafe_base64_encode('目标Bucket_Name:自定义文件key')
fops = fops+'|saveas/'+saveas_key
#任务类型:0: 普通任务 1: 闲时任务(一旦指定闲时任务,就不能指定 persistentPipeline)
persistentType = 0
#在上传策略中指定
policy={
'persistentOps': fops,
'persistentPipeline': pipeline,
'persistentType': persistentType
}
token = q.upload_token(bucket_name, key, 3600, policy)
localfile = './sync/bbb.jpg'
ret, info = put_file(token, key, localfile, version='v2')
print(info)
assert ret['key'] == key
assert ret['hash'] == etag(localfile)
Tips:上面的Demo只是针对视频转码功能,如果您需要使用比如音视频切片、视频截图、视频拼接等功能只需要修改上面 fops 后面的参数即可,如:fops = vframe/jpg/offset/1/w/480/h/360/rotate/90
就表示视频截图了。
可以看到上传成功后的行为主要是由上传凭证中的 上传策略 来指定。其中 上传策略 可以指定的行为不止这些,具体请参阅 上传策略 。
生成上传token
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
#需要填写你的 Access Key 和 Secret Key
access_key = ''
secret_key = ''
#构建鉴权对象
q = Auth(access_key, secret_key)
#要上传的空间
bucket_name = ''
#上传后保存的文件名
key = ''
#生成上传 Token,可以指定过期时间等
# 上传策略示例
# https://developer.qiniu.com/kodo/manual/1206/put-policy
policy = {
# 'callbackUrl':'https://requestb.in/1c7q2d31',
# 'callbackBody':'filename=$(fname)&filesize=$(fsize)'
# 'persistentOps':'imageView2/1/w/200/h/200'
}
#3600为token过期时间,秒为单位。3600等于一小时
token = q.upload_token(bucket_name, key, 3600, policy)
print(token)
上传加速
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu.services.storage.uploaders import FormUploader, ResumeUploaderV2
#需要填写你的 Access Key 和 Secret Key
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#构建鉴权对象
q = Auth(access_key, secret_key)
#要上传的空间
bucket_name = 'Bucket_Name'
#上传后保存的文件名
key = 'my-python-logo.png'
#生成上传 Token,可以指定过期时间等
token = q.upload_token(bucket_name, key, 3600)
#要上传文件的本地路径
localfile = './sync/bbb.jpg'
form_uploader = FormUploader(bucket_name, regions=[Region(up_host='https://<BucketId>.kodo-accelerate.<RegionId>.qiniucs.com')]) # 这里填写空间级别传输加速域名
ret, info = form_uploader.upload(key, localfile, up_token=token)
print(info, ret)
resumeUploader = ResumeUploaderV2(bucket_name, regions=[Region(up_host='https://<BucketId>.kodo-accelerate.<RegionId>.qiniucs.com')]) # 这里填写空间级别传输加速域名
ret, info = resumeUploader.upload(key, localfile, up_token=token)
print(info, ret)
持久化数据处理
触发持久化操作
如果需要对已存在于空间中的资源进行持久化数据处理(即触发持久化)。
以视频转码为例:
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth, PersistentFop, build_op, op_save, urlsafe_base64_encode
#对已经上传到存储的视频发起异步转码操作
access_key = 'Access_Key'
secret_key = 'Secret_Key'
q = Auth(access_key, secret_key)
#要转码的文件所在的空间和文件名。
bucket = 'Bucket_Name'
key = '1.mp4'
#转码是使用的队列名称。
pipeline = 'mpsdemo'
#要进行转码的转码操作。
fops = 'avthumb/mp4/s/640x360/vb/1.25m'
#可以对转码后的文件进行使用saveas参数自定义命名,当然也可以不指定文件会默认命名并保存在当前空间
saveas_key = urlsafe_base64_encode('目标Bucket_Name:自定义文件key')
fops = fops+'|saveas/'+saveas_key
force = 1
#任务类型:0: 普通任务 1: 闲时任务(一旦指定闲时任务,就不能指定 pipeline)
persistentType = 0
pfop = PersistentFop(q, bucket, pipeline)
ops = []
ops.append(fops)
ret, info = pfop.execute(key, ops, force, persistentType)
print(info)
assert ret['persistentId'] is not None
注意:
- 上面的Demo只是针对视频转码功能,如果您需要使用比如音视频切片、视频截图、视频拼接等功能只需要修改上面 fops 后面的参数即可,如:
fops = vframe/jpg/offset/1/w/480/h/360/rotate/90
就表示视频截图了。 - 闲时任务的功能介绍、使用场景、定价,详见 闲时任务策略说明。
但这个只是将转码这个耗时的操作提交到队列中,要想知道转码操作现在的状态, 需要根据返回的 persitentId 进行查询,查询接口。如果您不方便持续轮询每个异步处理的进度和状态,七牛可以异步处理完成后通知您们的业务服务器。这样就需要您在视频转码的例子中,初始化 PersistentFop 时添加上 notifyUrl , 来通知您们的业务服务器。
下载相关
生成时间戳防盗链
# -*- coding: utf-8 -*-
"""
获取一个配置时间戳防盗链的url
"""
from qiniu.services.cdn.manager import create_timestamp_anti_leech_url
import time
host = 'http://a.example.com'
# 配置时间戳时指定的key
encrypt_key = ''
# 资源路径
file_name = 'a/b/c/example.jpeg'
# 查询字符串,不需要加?
query_string = ''
# 截止日期的时间戳,秒为单位,3600为当前时间一小时之后过期
deadline = int(time.time())+3600
timestamp_url = create_timestamp_anti_leech_url(host, file_name, query_string, encrypt_key, deadline)
print(timestamp_url)
私有空间下载
# -*- coding: utf-8 -*-
# flake8: noqa
import requests
from qiniu import Auth
access_key = 'AK'
secret_key = 'SK'
q = Auth(access_key, secret_key)
#有两种方式构造base_url的形式
base_url = 'http://%s/%s' % (bucket_domain, key)
#或者直接输入url的方式下载
base_url = 'http://domain/key'
#可以设置token过期时间
private_url = q.private_download_url(base_url, expires=3600)
print(private_url)
r = requests.get(private_url)
assert r.status_code == 200
资源管理
资源管理包括的主要功能有:
- 获取文件信息
- 修改文件MimeType
- 修改文件存储类型
- 移动或重命名文件
- 复制文件副本
- 删除空间中的文件
- 设置或更新文件生存时间
- 获取指定前缀文件列表
- 抓取网络资源到空间
- 更新镜像存储空间中文件内容
- 资源管理批量操作
获取文件信息
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#初始化Auth状态
q = Auth(access_key, secret_key)
#初始化BucketManager
bucket = BucketManager(q)
#你要测试的空间, 并且这个key在你空间中存在
bucket_name = 'Bucket_Name'
key = 'python-logo.png'
#获取文件的状态信息
ret, info = bucket.stat(bucket_name, key)
print(info)
assert 'hash' in ret
获取指定前缀文件列表
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = 'Bucket_Name'
# 前缀
prefix = None
# 列举条目
limit = 10
# 列举出除'/'的所有文件以及以'/'为分隔的所有前缀
delimiter = None
# 标记
marker = None
ret, eof, info = bucket.list(bucket_name, prefix, marker, limit, delimiter)
print(info)
assert len(ret.get('items')) is not None
抓取网络资源到空间
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
bucket_name = 'Bucket_Name'
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
url = 'http://7xr875.com1.z0.glb.clouddn.com/xxx.jpg'
key = 'xxx.jpg'
ret, info = bucket.fetch(url, bucket_name, key)
print(info)
assert ret['key'] == key
抓取镜像源文件到空间
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
bucket_name = 'Bucket_Name'
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
# 要拉取的文件名
key = 'test.jpg'
ret, info = bucket.prefetch(bucket_name, key)
print(info)
assert ret['key'] == key
移动单个文件
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#初始化Auth状态
q = Auth(access_key, secret_key)
#初始化BucketManager
bucket = BucketManager(q)
#你要测试的空间, 并且这个key在你空间中存在
bucket_name = 'Bucket_Name'
key = 'python-logo.png'
#将文件从文件key 移动到文件key2,可以实现文件的重命名 可以在不同bucket移动
key2 = 'python-logo2.png'
ret, info = bucket.move(bucket_name, key, bucket_name, key2)
print(info)
assert ret == {}
复制单个文件
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#初始化Auth状态
q = Auth(access_key, secret_key)
#初始化BucketManager
bucket = BucketManager(q)
#你要测试的空间, 并且这个key在你空间中存在
bucket_name = 'Bucket_Name'
key = 'python-logo.png'
#将文件从文件key 复制到文件key2。 可以在不同bucket复制
key2 = 'python-logo2.png'
ret, info = bucket.copy(bucket_name, key, bucket_name, key2)
print(info)
assert ret == {}
删除单个文件
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = 'Access_Key'
secret_key = 'Secret_Key'
#初始化Auth状态
q = Auth(access_key, secret_key)
#初始化BucketManager
bucket = BucketManager(q)
#你要测试的空间, 并且这个key在你空间中存在
bucket_name = 'Bucket_Name'
key = 'python-logo.png'
#删除bucket_name 中的文件 key
ret, info = bucket.delete(bucket_name, key)
print(info)
assert ret == {}
设置或更新文件生存时间
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
#初始化Auth状态
q = Auth(access_key, secret_key)
#初始化BucketManager
bucket = BucketManager(q)
#你要测试的空间, 并且这个key在你空间中存在
bucket_name = 'Bucket_Name'
key = 'python-test.png'
#您要更新的生命周期
days = '5'
ret, info = bucket.delete_after_days(bucket_name, key, days)
print(info)
修改文件存储类型
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = 'Bucket_Name'
key = '...'
ret, info = bucket.change_type(bucket_name, key, 1) # 0 是标准存储,1 表示低频存储,2 表示归档存储,3 表示深度归档存储,4 表示归档直读存储
print(info)
修改资源元信息
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager
access_key = '...'
secret_key = '...'
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = 'Bucket_Name'
key = '...'
# 将一个文件的元信息修改为jpg
ret, info = bucket.change_mime(bucket_name, key, 'image/jpg')
print(info)
批量操作
# -*- coding: utf-8 -*-
# flake8: noqa
from qiniu import Auth
from qiniu import BucketManager,build_batch_rename
from qiniu import build_batch_copy
from qiniu import build_batch_move
access_key = '...'
secret_key = '...'
# 初始化Auth状态
q = Auth(access_key, secret_key)
# 初始化BucketManager
bucket = BucketManager(q)
keys = {'123.jpg':'123.jpg'}
# ops = build_batch_copy(bucket_name, keys, 'teest',force='true')
# ops = build_batch_move(bucket_name, keys, 'teest', force='true')
ops = build_batch_rename(bucket_name, keys,force='true')
ret, info = bucket.batch(ops)
print(ret)
print(info)
assert ret == {}
批量查询文件信息
# -*- coding: utf-8 -*-
"""
批量查询文件信息
https://developer.qiniu.com/kodo/api/1250/batch
"""
from qiniu import build_batch_stat, Auth, BucketManager
access_key = ''
secret_key = ''
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = ''
# 需要查询的文件名
keys = ['1.gif', '2.txt', '3.png', '4.html']
ops = build_batch_stat(bucket_name, keys)
ret, info = bucket.batch(ops)
print(info)
批量重命名文件
# -*- coding: utf-8 -*-
"""
批量重命名文件
https://developer.qiniu.com/kodo/api/1250/batch
"""
from qiniu import build_batch_rename, Auth, BucketManager
access_key = ''
secret_key = ''
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = ''
# force为true时强制同名覆盖, 字典的键为原文件,值为目标文件
ops = build_batch_rename(bucket_name, {'src_key1': 'target_key1', 'src_key2': 'target_key2'}, force='true')
ret, info = bucket.batch(ops)
print(info)
批量移动文件
# -*- coding: utf-8 -*-
"""
批量移动文件
https://developer.qiniu.com/kodo/api/1250/batch
"""
from qiniu import build_batch_move, Auth, BucketManager
access_key = ''
secret_key = ''
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
src_bucket_name = ''
target_bucket_name = ''
# force为true时强制同名覆盖, 字典的键为原文件,值为目标文件
ops = build_batch_move(src_bucket_name, {'src_key1': 'target_key1', 'src_key2': 'target_key2'}, target_bucket_name, force='true')
ret, info = bucket.batch(ops)
print(info)
批量删除文件
# -*- coding: utf-8 -*-
"""
批量删除文件
https://developer.qiniu.com/kodo/api/1250/batch
"""
from qiniu import build_batch_delete, Auth, BucketManager
access_key = ''
secret_key = ''
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
bucket_name = ''
keys = ['1.gif', '2.txt', '3.png', '4.html']
ops = build_batch_delete(bucket_name, keys)
ret, info = bucket.batch(ops)
print(info)
批量复制文件
# -*- coding: utf-8 -*-
"""
批量拷贝文件
https://developer.qiniu.com/kodo/api/1250/batch
"""
from qiniu import build_batch_copy, Auth, BucketManager
access_key = ''
secret_key = ''
q = Auth(access_key, secret_key)
bucket = BucketManager(q)
src_bucket_name = ''
target_bucket_name = ''
# force为true时强制同名覆盖, 字典的键为原文件,值为目标文件
ops = build_batch_copy(src_bucket_name, {'src_key1': 'target_key1', 'src_key2': 'target_key2'}, target_bucket_name, force='true')
ret, info = bucket.batch(ops)
print(info)
CDN相关
刷新节点资源
# -*- coding: utf-8 -*-
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = '...'
secret_key = '...'
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
# 需要刷新的文件链接
urls = [
'http://aaa.example.com/a.gif',
'http://bbb.example.com/b.jpg'
]
# 刷新链接
refresh_url_result = cdn_manager.refresh_urls(urls)
print(refresh_url_result)
刷新整个目录下的节点资源
# -*- coding: utf-8 -*-
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = '...'
secret_key = '...'
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
# 需要刷新的目录链接
dirs = [
'http://aaa.example.com/doc/img/',
'http://bbb.example.com/doc/video/'
]
# 刷新链接
refresh_dir_result = cdn_manager.refresh_dirs(dirs)
预取资源到cdn结点
# -*- coding: utf-8 -*-
"""
预取资源到cdn节点
https://developer.qiniu.com/fusion/api/1227/file-prefetching
"""
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = '...'
secret_key = '...'
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
# 需要刷新的文件链接
urls = [
'http://aaa.example.com/doc/img/',
'http://bbb.example.com/doc/video/'
]
# 刷新链接
refresh_dir_result = cdn_manager.prefetch_urls(urls)
获取指定域名指定时间内的日志链接
# -*- coding: utf-8 -*-
"""
获取指定域名指定时间内的日志链接
"""
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = ''
secret_key = ''
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
log_date = '2017-07-20'
urls = [
'a.example.com',
'b.example.com'
]
ret, info = cdn_manager.get_log_list_data(urls, log_date)
print(ret)
print(info)
获取指定域名指定时间段内的流量
# -*- coding: utf-8 -*-
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = ''
secret_key = ''
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
startDate = '2017-07-20'
endDate = '2017-08-20'
granularity = 'day'
urls = [
'a.example.com',
'b.example.com'
]
# 获得指定域名流量
ret, info = cdn_manager.get_flux_data(urls, startDate, endDate, granularity)
print(ret)
print(info)
获取指定域名指定时间段内的带宽
# -*- coding: utf-8 -*-
import qiniu
from qiniu import CdnManager
# 账户ak,sk
access_key = ''
secret_key = ''
auth = qiniu.Auth(access_key=access_key, secret_key=secret_key)
cdn_manager = CdnManager(auth)
startDate = '2017-07-20'
endDate = '2017-08-20'
granularity = 'day'
urls = [
'a.example.com',
'b.example.com'
]
ret, info = cdn_manager.get_bandwidth_data(urls, startDate, endDate, granularity)
print(ret)
print(info)
API参考手册
常见问题
- 第二个参数 info 保留了请求响应的信息,失败情况下 ret 为 none ,将 info 打印出来,提交给我们。
- API 的使用 demo 可以参考 单元测试。
- 如果碰到
ImportError: No module named requests.auth
请安装requests
。
相关资源
如果您有任何关于我们文档或产品的建议和想法,欢迎到我们的技术论坛参与讨论。
- 技术论坛 - 在这里您可以和其他开发者愉快的讨论如何更好的使用七牛云服务
- 提交工单 - 如果您的问题不适合在论坛讨论或得不到回答,您也可以提交一个工单,技术支持人员会尽快回复您
- 博客 - 这里会持续发布市场活动和技术分享文章
- 微博
- 常见问题
贡献代码
-
Fork
-
创建您的特性分支 git checkout -b my-new-feature
-
提交您的改动 git commit -am ‘Added some feature’
-
将您的修改记录提交到远程 git 仓库 git push origin my-new-feature
-
然后到 github 网站的该 git 远程仓库的 my-new-feature 分支下发起 Pull Request
许可证
Copyright © 2014 qiniu.com
基于 MIT 协议发布: