Python SDK

pip install qiniu
或
easy_install qiniu

#从Python SDK 下载地址下载源码
tar xvzf python-sdk-$VERSION.tar.gz
cd python-sdk-$VERSION
python setup.py install

from qiniu import Auth

q = Auth(access_key, secret_key)

# -*- coding: utf-8 -*-
# flake8: noqa

from qiniu import Auth, put_file_v2, 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_v2(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_v2, 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_v2(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_v2, 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）
persistent_type = 0
#任务模版 ID，使用预设好的任务；与 fops 冲突，只能二选一
#persistent_workflow_template_id = 'your-workflow-id'

#在上传策略中指定
policy={
  'persistentOps': fops,
  'persistentPipeline': pipeline,
  'persistentType': persistent_type
  #'persistentWorkflowTemplateID': persistent_workflow_template_id
 }

token = q.upload_token(bucket_name, key, 3600, policy)

localfile = './sync/bbb.jpg'

ret, info = put_file_v2(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

#需要填写你的 Access Key 和 Secret Key
access_key = ''
secret_key = ''

#构建鉴权对象
q = Auth(access_key, secret_key)

#要上传的空间
bucket_name = ''

#上传后保存的文件名
key = ''

#生成上传 Token，可以指定过期时间等

# 上传策略示例
# /products/kodo/development-guidelines/security/1-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, Region
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）
persistent_type = 0
#任务模版 ID，使用预设好的任务；与 fops 冲突，只能二选一
#workflow_template_id = 'your-workflow-id'

pfop = PersistentFop(q, bucket, pipeline)
ops = []
ops.append(fops)
ret, info = pfop.execute(
  key,
  ops=ops,
  #workflow_template_id=workflow_template_id
  force=force,
  persistent_type=persistent_type
)
print(info)
assert ret['persistentId'] is not None

# -*- 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

# -*- 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 -*-
"""
批量查询文件信息

/products/kodo/rs/0-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 -*-
"""
批量重命名文件

/products/kodo/rs/0-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 -*-
"""
批量移动文件

/products/kodo/rs/0-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 -*-
"""
批量删除文件

/products/kodo/rs/0-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 -*-
"""
批量拷贝文件

/products/kodo/rs/0-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)

# -*- 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)

# -*- coding: utf-8 -*-
"""
预取资源到cdn节点

/faq/products/cdn/6-the-significance-of-file-prefetching-and-operating-methods
"""

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)