最近更新时间:2024-11-27 19:30:08
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
#遍历key,若是文件打印文件名,若是目录打印目录
keys = b.list()
for k in keys:
if isinstance(k, Key):
print('文件:%s' % k.name)
elif isinstance(k, Prefix):
print('目录:%s' % k.name)
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 列举 images 文件夹下的所有文件。比如b.list(prefix="images/")
keys = b.list(prefix="<yourPrefix>")
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# delimiter 为对文件名称进行分组的字符;delimiter 为空时,默认为'/'
# 返回结果中 delimiter 分隔符之前的字符会放入 commonPrefixes 中,可以类比理解为文件夹
# 只列举该文件夹下的文件和子文件夹(目录)名称,子文件夹下的文件和文件夹不显示。
keys = b.list(delimiter='/')
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 列举指定字符串(object名称)之后(字典排序)的所有文件。同名 object 也会被排除。
keys = b.list(marker='<yourMarker>')
Python SDK的版本V1.0.19开始支持max_keys参数
from ks3.connection import Connection
from itertools import islice
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 如下列举10个文件
keys = islice(b.list(max_keys=10), 10)
for k in keys:
print(k.name)
Python SDK的版本V1.11.0开始支持retry_policy、request_interval参数
list方法会自动调用多次list接口,直到文件被列举完或达到指定个数。配置重试策略可缓解列举中途因网络等问题导致的失败。
from ks3.connection import Connection
from ks3.utils import RetryPolicy, ExponentialWait, StopAfterAttempt, JitterWait, StopAfterDelay
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 配置重试策略
# 默认尝试3次,第一次等待1.5秒,第二次等待3秒
# ExponentialWait 指数退避时间
# wait_time = multiplier * (2 ** (attempt_number - 1)),其中attempt_number从1开始
# wait_time 的范围为 [min_interval, max_interval]
# StopAfterAttempt 达到最大重试次数后停止重试。
retry_policy = RetryPolicy(
wait_strategy=ExponentialWait(multiplier=1.5, min_interval=1, max_interval=10),
stop_strategy=StopAfterAttempt(max_attempts=3)
)
# 该配置的效果为:最多调用3次,第一次失败后等待1.5秒再进行第二次尝试,第二次失败后等待3秒再进行第三次尝试。
keys = b.list(retry_policy=retry_policy)
for k in keys:
print(k.name)
# JitterWait 抖动退避时间
# wait_time = initial + (random.uniform(0, jitter))
# wait_time 的范围为 [initial, max_interval]
# StopAfterDelay 从第一次调用开始计时,经过指定时间后停止重试。
retry_policy = RetryPolicy(
wait_strategy=JitterWait(initial=1, jitter=5, max_interval=10),
stop_strategy=StopAfterDelay(max_delay=10)
)
# 该配置的效果为:每次失败后在[initial, max_interval]范围内等待随机时间后再进行下一次尝试,经过10秒后停止重试。
keys = b.list(retry_policy=retry_policy)
for k in keys:
print(k.name)
# 配置每次成功请求之间的等待时间
# 默认等待时间为0
keys = b.list(request_interval=1)
for k in keys:
print(k.name)
Python SDK的版本V1.3.2开始支持时间过滤参数start_time, end_time
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 列举 Bucket 内指定前缀的文件以及指定时间区间的文件
# start_time = 1625460400 这个时间之前的数据
# end_time = 1625460400 这个时间之后的数据
# start_time=1640331446, end_time=1641895096 这个时间中间的数据
keys = b.listObjects(prefix="local/load", start_time=1640331446, end_time=1641895096)
from ks3.connection import Connection
from ks3.prefix import Prefix
from ks3.key import Key
from ks3.prefix import Prefix
# 金山云主账号 AccessKey 拥有所有API的访问权限,风险很高。
# 强烈建议您创建并使用子账号进行 API 访问或日常运维,请登录 https://uc.console.ksyun.com/pro/iam/#/user/list 创建子账号。
# 通过指定 host(Endpoint),您可以在指定的地域创建新的存储空间。host(Endpoint) 以北京为例,其它 Region 请按实际情况填写。
conn = Connection('<yourAccessKeyId>', '<yourAccessKeySecret>', host='ks3-cn-beijing.ksyuncs.com')
# 获取存储空间实例
b = conn.get_bucket('<yourBucketName>')
# 计算指定目录下的文件大小
def calcFolderSize(bucket, folder):
length = 0
# 列举文件夹下的所有文件,包括子目录下的文件。
keys = bucket.list(prefix=folder)
# 计算文件大小
for k in keys:
length += k.size
return length
def getFolderSizeInBucket():
# 列举该文件夹下的文件和子文件夹(目录)名称。
keys = b.list(delimiter='/')
for k in keys:
# 判断是否为目录
if isinstance(k, Prefix):
print('dir: ' + k.name + ' size:' + str(calcFolderSize(b, k.name)) + "Byte")
纯净模式
鼠标选中内容,快速反馈问题