• 热门
  • 基础
  • 数据库
  • 安全
  • 大数据
  • 人工智能
  • 混合云
  • 开发与运维
  • 企业应用

应用服务

行业引擎

全部文档
当前文档

暂无内容

如果没有找到您期望的内容,请尝试其他搜索词

文档中心

列举文件(Python)

最近更新时间:2024-11-27 19:30:08

列举 Bucket 内的文件或者目录

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)

列举 Bucket 内指定前缀的文件

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>")

列举 Bucket 内以指定分隔符分组的文件

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)

列举 Bucket 内指定前缀的文件以及指定时间区间的文件

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")

纯净模式常规模式

纯净模式

点击可全屏预览文档内容

鼠标选中内容,快速反馈问题

如果在文档使用中出现问题,可选中有问题的部分进行快速反馈,我们将跟进处理。
不再提示
好的,我知道了

聆听反馈