pip install six
pip install ks3sdk
通过git下载SDK到本地
进入ks3-python-sdk目录
cd ks3-python-sdk
安装SDK
python setup.py install
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
参数说明
ak:金山云提供的ACCESS KEY ID
sk:金山云提供的SECRET KEY ID
host:金山云提供的各个Region的域名(例 ks3-cn-beijing.ksyun.com),具体定义可参考 API接口文档-Region(区域)。也可以是用户自定义域名,如果是用户自定义域名,需要将domain_mode设置为True。
is_secure:是否通过HTTPS协议访问Ks3,True:启用 False:关闭。
domain_mode:是否使用自定义域名访问Ks3(host填写自定义域名),True:是 False:否。
适用于2.6、2.7、3.3、3.4、3.5、3.6、3.7版本。
列出客户所有的 Bucket 信息
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
buckets = c.get_all_buckets()
for b in buckets:
print(b.name)
创建一个新的Bucket
在建立了连接后,可以创建一个bucket。bucket在s3中是一个用于储存key/value的容器。用户可以将所有的数据存储在一个bucket里,也可以为不同种类数据创建相应的bucket。
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.create_bucket(bucket_name)
注:这里如果出现409 conflict错误,说明请求的bucket name有冲突,因为bucket name是全局唯一的
删除指定Bucket
删除一个bucket可以通过delete_bucket方法实现。
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
c.delete_bucket(bucket_name)
如果bucket下面存在key,那么需要首先删除所有key
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
for k in b.list():
k.delete()
c.delete_bucket(bucket_name)
获取Bucket的ACL
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
policy = b.get_acl()
for grant in policy.acl.grants:
print(grant.permission, grant.display_name, grant.email_address, grant.id)
设置Bucket的ACL
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
#设置bucket的权限, private or public-read or public-read-write
b.set_acl("public-read")
为Bucket设置跨区域复制规则
对bucket配置跨region复制规则,支持添加、查询、删除
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
bucket = c.get_bucket(bucket_name)
bucket.set_bucket_crr('emmacrra', 'Disabled')
#bucket_crr = bucket.get_bucket_crr()
#bucket.delete_bucket_crr()
获取Object元信息
获取Object元数据信息(大小、最后更新时间等)
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
b = c.get_bucket(bucket_name)
try:
k = b.get_key(key_name)
if k:
print(k.name, k.size, k.last_modified)
except:
pass #异常处理
下载该Object数据
下载object,并且作为字符串返回
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
b = c.get_bucket(bucket_name)
try:
k = b.get_key(key_name)
s = k.get_contents_as_string().decode()
print(s)
except:
pass #异常处理
下载object,并且保存到文件中
#保存到文件
k.get_contents_to_filename("SAVED_FILE_PATH")
上传Object数据
将指定目录下某一个文件上传,同时可以指定文件ACL
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
try:
b = c.get_bucket(bucket_name)
k = b.new_key(key_name)
#object policy : 'private' or 'public-read'
ret=k.set_contents_from_filename("YOUR_SOURCE_FILE_PATH", policy="private")
if ret and ret.status == 200:
print("上传成功")
except:
pass #异常处理
将字符串作为value上传
k.set_contents_from_string('YOUR_FILE_CONTENTS')
上传回调
需要设置x-kss-callbackurl
和x-kss-callbackbody
请求头,详见文档上传回调处理。
headers = {"x-kss-callbackurl": "Callback_URL","x-kss-callbackbody":"objectKey=${key}&etag=${etag}&uid=123"}
ret = k.set_contents_from_filename(filepath, headers=headers)
以低频存储方式上传
//x-kss-storage-class有效值为"STANDARD"、"STANDARD_IA"。"STANDARD"表示标准存储,"STANDARD_IA"表示低频存储,如果不指定,默认为标准存储。
headers = {"x-kss-storage-class": "STANDARD_IA"}
resp = k.set_contents_from_filename(filepath, policy="private", headers=headers)
抓取网络资源到KS3
content-type
为非必填
x-kss-acl
根据需要设定公开读 public-read,当未设置时为私密
import ks3.auth as auth
class TestFetchObject(unittest.TestCase):
# Create a object key in specified bucket
def test_fetch_object(self):
auth.qsa_of_interest.append('fetch')
headerss = {'x-kss-sourceurl': 'http%3A%2F%2Fpic1.sc.chinaz.com%2Ffiles%2Fpic%2Fpic9%2F202006%2Fbpic20469.jpg',
'x-kss-acl': 'public-read', 'Content-Type': 'application/xml'}
resp = conn.make_request('PUT', test_bucket, "testfetch", query_args='fetch', headers=headerss)
print(resp.read())
复制Object数据
将指定Bucket下的文件复制到本Bucket下(需要对源Bucket下的文件具有读权限)
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_DST_BUCKET_NAME"
key_name = "YOUR_DST_KEY_NAME"
try:
b = c.get_bucket(bucket_name)
b.copy_key(key_name,"YOUR_SRC_BUCKET_NAME","YOUR_SRC_KEY_NAME")
except:
pass #异常处理
删除Object数据
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
try:
b=c.get_bucket(bucket_name)
b.delete_key(key_name)
except:
pass #异常处理
列举Bucket内的文件或者目录
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
keys = b.list(delimiter='/')
for k in keys:
if isinstance(k, Key):
print('file:%s' % k.name)
elif isinstance(k, Prefix):
print('dir:%s' % k.name)
列举Bucket内指定前缀的文件
keys = b.list(prefix="PREFIX")
获得Object的acl
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
b = c.get_bucket(bucket_name)
policy = b.get_acl(key_name)
print(policy.to_xml())
#object policy : 'private' or 'public-read'
b.set_acl("public-read", test_key)
分块上传
如果你想上传一个大文件,你可以将它分成几个小份,逐个上传,ks3会按照顺序把它们合成一个最终的object。通过SDK进行分块上传能够保障数据传到ks3的正确性,无需另外做数据校验。
整个过程需要几步来完成,下面的demo程序是通过python的FileChunkIO模块来实现的。所以可能需要首先运行pip install FileChunkIO来安装。
import math, os
from ks3.connection import Connection
from filechunkio import FileChunkIO
# Connect to S3
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
# Get file info
source_path = 'SOURCE_FILE_PATH'
source_size = os.stat(source_path).st_size
# Create a multipart upload request
# 此处os.path.basename(source_path)可以替换为需要设置的objectKey
mp = b.initiate_multipart_upload(os.path.basename(source_path), policy="private")
# Use a chunk size of 50 MiB (feel free to change this)
chunk_size = 52428800
chunk_count = int(math.ceil(source_size*1.0 / chunk_size*1.0))
# Send the file parts, using FileChunkIO to create a file-like object
# that points to a certain byte range within the original file. We
# set bytes to never exceed the original file size.
try:
for i in range(chunk_count):
offset = chunk_size * i
bytes = min(chunk_size, source_size - offset)
with FileChunkIO(source_path, 'r', offset=offset, bytes=bytes) as fp:
mp.upload_part_from_file(fp, part_num=i + 1)
ret = mp.complete_upload()
if ret and ret.status == 200:
print("上传成功")
except:
pass # 异常处理
说明
以低频存储方式上传,需要在initiate_multipart_upload阶段设置
//x-kss-storage-class有效值为"STANDARD"、"STANDARD_IA"。"STANDARD"表示标准存储,"STANDARD_IA"表示低频存储,如果不指定,默认为标准存储。
headers = {"x-kss-storage-class": "STANDARD_IA"}
mp = b.initiate_multipart_upload(os.path.basename(source_path), policy="private", headers=headers)
multi_task_result = {}
def multi_put():
import threadpool
conn = Connection("AK","SK",host="ENDPOINT")
b = conn.get_bucket("BUCKETNAME")
ks3_key=""
local_path=""
retry_times=3
thread_num=5
f_size = os.stat(local_path).st_size
mp = b.initiate_multipart_upload(ks3_key)
if not mp:
raise RuntimeError("%s init multiupload error" % ks3_key)
print "%s begin multipart upload,uploadid=%s" % (ks3_key, mp.id)
chunk_size = 104857600
chunk_count = int(math.ceil(f_size / float(chunk_size)))
pool_args_list = []
try:
for i in range(chunk_count):
offset = chunk_size * i
bs = min(chunk_size, f_size - offset)
part_num = i + 1
pool_args_list.append(
([mp, local_path, ks3_key, offset, bs, part_num, retry_times], None))
pool = threadpool.ThreadPool(thread_num)
requests = threadpool.makeRequests(upload_part_task, pool_args_list)
[pool.putRequest(req) for req in requests]
pool.wait()
global multi_task_result
if len(multi_task_result) != chunk_count:
raise RuntimeError(
"%s part miss,expect=%d,actual=%d" % (ks3_key, chunk_count, len(multi_task_result)))
for item in multi_task_result.keys():
if not multi_task_result[item]:
raise RuntimeError("%s part upload has fail" % ks3_key)
mp.complete_upload()
print "%s multipart upload success" % ks3_key
except BaseException, e:
print "%s multipart upload fail" % ks3_key
if mp:
mp.cancel_upload()
raise e
def upload_part_task(mp, local_path, ks3_key, offset, chunk_size, part_num, retry_times):
global multi_task_result
cur_task_ret = False
try:
for i in range(retry_times):
try:
with FileChunkIO(local_path, 'rb', offset=offset, bytes=chunk_size) as fp:
mp.upload_part_from_file(fp, part_num=part_num)
cur_task_ret = True
print "%s part %d upload success" % (ks3_key, part_num)
break
except BaseException, e:
print "%s part %d fail,uploadid=%s,error=%s" % (
ks3_key, part_num, mp.id, traceback.format_exc())
if i + 1 >= retry_times:
print "%s part %d upload fail" % (ks3_key, part_num)
raise e
except BaseException, e:
cur_task_ret = False
finally:
multi_task_result[part_num] = cur_task_ret
获取已上传分块列表
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
b = c.get_bucket(bucket_name)
for p in b.list_multipart_uploads():
print('uploadId:%s,key:%s' % (p.id, p.key_name))
for i in p:
print(i.part_number, i.size, i.etag, i.last_modified)
生成下载外链地址
对私密属性的文件生成下载外链(该链接具有时效性)
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
key_name = "YOUR_KEY_NAME"
b = c.get_bucket(bucket_name)
k = b.get_key(key_name)
if k:
url = k.generate_url(60) # 60s 后该链接过期
print(url)
指定时间过期
k.generate_url(1492073594,expires_in_absolute=True) # 1492073594为Unix Time
修改文件元数据
class ModifyObjectMeta(unittest.TestCase):
#srcbucket == dstbucket, srckey == dstkey, x-kss-metadata-directive=REPLACE
def modify_object_meta(self):
b = conn.get_bucket(test_bucket)
b.copy_key('1copykey.jpg', test_bucket, '1copykey.jpg', headers={'content-type':'text/plain', 'x-kss-metadata-directive':'REPLACE'})
修改文件存储类型
class ModifyStorageClass(unittest.TestCase):
# srcbucket == dstbucket, srckey == dstkey, x-kss-storage-class=new storage class
def modify_storage_class(self):
b = conn.get_bucket(test_bucket)
b.copy_key('1copykey.jpg', test_bucket, '1copykey.jpg', headers={'x-kss-storage-class':'STANDARD_IA'})
桶空间策略管理
对bucket进行空间策略配置,支持添加、查询、删除
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
bucket = c.get_bucket(bucket_name)
bucket.set_bucket_policy(
policy='{"Statement":[{"Resource":["krn:ksc:ks3:::jiangran123","krn:ksc:ks3:::jiangran123/*"],"Principal":{"KSC":["krn:ksc:iam::32432423:root"]},"Action":["ks3:*"],"Effect":"Allow"}]}')
#policy = bucket.get_bucket_policy()
#bucket.delete_bucket_policy()
归档Object解冻
此接口用于对归档Object进行解冻
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False)
bucket_name = "YOUR_BUCKET_NAME"
bucket = c.get_bucket(bucket_name)
restore_result = bucket.restore_object(test_key)
如果您有上传前先加密数据的需求,可以考虑使用加密模式。
使用加密客户端需要安装pycrypto。
您可以使用自己的密钥文件,或者选择调用我们的密钥生成方法。
from ks3.encryption import Crypts
Crypts.generate_key('your_path', 'key_name')
请注意保管好您的key文件,KS3服务端将不会对客户端加密时使用的key文件进行保存,一旦丢失,文件将无法被解密。
对put、get方法加密客户端和普通客户端用法基本一致,不同之处在于初始化Connection对象时需要多填两个参数:local_encrypt和local_key_path。local_encrypt设置为true,local_key_path设置为密钥存放在本地的地址。
from ks3.connection import Connection
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT',is_secure=False, domain_mode=False, local_encrypt=True,
local_key_path="your_key_path")
b = c.get_bucket("your_bucket")
#put
kw = b.new_key("your_key")
ret = kw.set_contents_from_string("some string")
#get
get_key = b.get_key("your_key")
s = get_key.get_contents_as_string()
print("Result:",s)
注意:加密客户端的分块上传不支持对文件的并行上传!分块上传时必须依次序上传每一块,否则数据将无法解密。
示例1:使用FileChunkIO进行分块上传。与普通客户端的方法基本一致,和put一样只需在初始化时增加参数。
c = Connection(ak, sk, host,is_secure=False, domain_mode=False, local_encrypt=True, local_key_path="your_key_path")
#继续普通分块上传方法
示例2:自己切分文件进行分块上传。除了修改Connection的参数之外,还需在调用upload_part_from_file方法时指定is_last_part的值。
from ks3.connection import Connection
from filechunkio import FileChunkIO
import math, os
ak = 'YOUR_ACCESS_KEY'
sk = 'YOUR_SECRET_KEY'
c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT',is_secure=False, domain_mode=False, local_encrypt=True,
local_key_path="your_key_path")
b = c.get_bucket("your_bucket")
source_path = 'SOURCE_FILE_PATH'
source_size = os.stat(source_path).st_size
mp = bucket.initiate_multipart_upload(os.path.basename(source_path))
# Use a chunk size of 50 MiB (feel free to change this)
chunk_size = 52428800
chunk_count = int(math.ceil(source_size*1.0 / chunk_size*1.0))
try:
for i in range(chunk_count):
offset = chunk_size * i
last = False
bytes = min(chunk_size, source_size - offset)
if i +1 == chunk_count:
last = True
with FileChunkIO(source_path, 'r', offset=offset,bytes=bytes) as fp:
mp.upload_part_from_file(fp, part_num=i + 1, is_last_part=last)
ret = mp.complete_upload()
if ret and ret.status == 200:
print("上传成功")
except:
pass
注意事项
- 对于使用加密模式上传的数据,请使用加密模式下(local_encrypt=True)的get方法进行下载。未设置加密模式的get下载下来的这份数据是加密的,无法解读。
- 加密上传默认进行MD5验证,以防止网络传输过程中的数据损坏。在文件较大的情况下,对加密后文件的MD5计算较为耗时(每500MB约耗时10s),如果不能接受这种额外耗时,可以在调用方法时设置calc_md5=False来关闭MD5校验功能。当然,我们不推荐您关闭MD5校验。
#PUT上传取消MD5 CHECK:
ret = kw.set_contents_from_string(test_str, calc_encrypt_md5=False)
ret = kw.set_contents_from_filename(test_path, calc_encrypt_md5=False)
#分块上传时取消MD5 CHECK:
mp = b.initiate_multipart_upload(os.path.basename(path), calc_encrypt_md5=False)
import hashlib
import base64
md5_generator = hashlib.md5()
md5_generator.update("YOUR_KEY".encode())
print(base64.b64encode(md5_generator.hexdigest().encode()))
文档内容是否对您有帮助?
评价建议不能为空
非常感谢您的反馈,我们会继续努力做到更好!