如何通过Python SDK V2管理Bucket的清单配置_对象存储(OSS)-阿里云帮助中心

本文介绍如何通过Python SDK V2添加、查看、批量列举和删除存储空间（Bucket）的清单（Inventory）配置。

注意事项

本文示例代码以华东1（杭州）的地域IDcn-hangzhou为例，默认使用外网Endpoint，如果您希望通过与OSS同地域的其他阿里云产品访问OSS，请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系，请参见OSS地域和访问域名。
请确保您拥有调用添加、查看、列举和删除存储空间清单配置的权限。Bucket所有者默认拥有此类权限，如果您无此类权限，请先向Bucket所有者申请对应操作的权限。
单个Bucket最多只能有1000条清单配置。
配置清单的源Bucket与存放导出的清单文件所在的目标Bucket必须位于同一个Region。

示例代码

添加清单配置

以下代码用于为某个Bucket添加清单配置：

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，并描述脚本用途：设置存储空间清单（Inventory）
parser = argparse.ArgumentParser(description="put bucket inventory sample")

# 定义命令行参数，包括必需的区域、存储空间名称、endpoint、用户ID、角色ARN以及清单名称
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--user_id', help='User account ID.', required=True)
parser.add_argument('--arn', help='The Alibaba Cloud Resource Name (ARN) of the role that has the permissions to read all objects from the source bucket and write objects to the destination bucket. Format: `acs:ram::uid:role/rolename`.', required=True)
parser.add_argument('--inventory_id', help='The name of the inventory.', required=True)

def main():
    # 解析命令行参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 发送请求以配置指定存储空间的清单设置
    result = client.put_bucket_inventory(oss.PutBucketInventoryRequest(
            bucket=args.bucket,  # 存储空间名
            inventory_id=args.inventory_id,  # 存储空间清单ID
            inventory_configuration=oss.InventoryConfiguration(
                included_object_versions='All',  # 包含所有版本的对象
                optional_fields=oss.OptionalFields(
                    fields=[  # 可选字段，如大小和最后修改日期
                        oss.InventoryOptionalFieldType.SIZE,
                        oss.InventoryOptionalFieldType.LAST_MODIFIED_DATE,
                    ],
                ),
                id=args.inventory_id,  # 存储空间清单ID
                is_enabled=True,  # 启用存储空间清单
                destination=oss.InventoryDestination(
                    oss_bucket_destination=oss.InventoryOSSBucketDestination(
                        format=oss.InventoryFormatType.CSV,  # 输出格式为CSV
                        account_id=args.user_id,  # 用户账户ID
                        role_arn=args.arn,  # 角色ARN，具有读取源存储空间和写入目标存储空间的权限
                        bucket=f'acs:oss:::{args.bucket}',  # 目标存储空间
                        prefix='aaa',  # 清单文件前缀
                    ),
                ),
                schedule=oss.InventorySchedule(
                    frequency=oss.InventoryFrequencyType.DAILY,  # 清单频率，这里设置为每天
                ),
                filter=oss.InventoryFilter(
                    lower_size_bound=1024,  # 对象大小下限（字节）
                    upper_size_bound=1048576,  # 对象大小上限（字节）
                    storage_class='ColdArchive',  # 存储类别筛选条件
                    prefix='aaa',  # 对象前缀筛选条件
                    last_modify_begin_time_stamp=1637883649,  # 最后修改时间戳开始范围
                    last_modify_end_time_stamp=1638347592,  # 最后修改时间戳结束范围
                ),
            ),
    ))

    # 打印操作结果的状态码和请求ID，以便确认请求状态
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始

查看清单配置

以下代码用于查看某个Bucket的清单配置：

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，并描述脚本用途：获取存储空间清单配置（Inventory）
parser = argparse.ArgumentParser(description="get bucket inventory sample")

# 定义命令行参数，包括必需的区域、存储空间名称、endpoint以及清单ID
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--inventory_id', help='The name of the inventory.', required=True)

def main():
    # 解析命令行参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 发送请求以获取指定存储空间的清单配置详情
    result = client.get_bucket_inventory(oss.GetBucketInventoryRequest(
            bucket=args.bucket,  # 存储空间名
            inventory_id=args.inventory_id,  # 清单配置ID
    ))

    # 打印操作结果的状态码和请求ID，以便确认请求状态
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' included object versions: {result.inventory_configuration.included_object_versions},'
          f' id: {result.inventory_configuration.id},'
          f' is enabled: {result.inventory_configuration.is_enabled},'
          f' account id: {result.inventory_configuration.destination.oss_bucket_destination.account_id},'
          f' role arn: {result.inventory_configuration.destination.oss_bucket_destination.role_arn},'
          f' bucket: {result.inventory_configuration.destination.oss_bucket_destination.bucket},'
          f' prefix: {result.inventory_configuration.destination.oss_bucket_destination.prefix},'
          # 注释掉的这两行是关于加密配置的信息，如果需要可以取消注释
          # f' key id: {result.inventory_configuration.destination.oss_bucket_destination.encryption.sse_kms.key_id},'
          # f' sse oss: {result.inventory_configuration.destination.oss_bucket_destination.encryption.sse_oss},'
          f' lower size bound: {result.inventory_configuration.filter.lower_size_bound},'
          f' upper size bound: {result.inventory_configuration.filter.upper_size_bound},'
          f' storage class: {result.inventory_configuration.filter.storage_class},'
          f' prefix: {result.inventory_configuration.filter.prefix},'
          f' last modify begin time stamp: {result.inventory_configuration.filter.last_modify_begin_time_stamp},'
          f' last modify end time stamp: {result.inventory_configuration.filter.last_modify_end_time_stamp},'
    )

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始

列举清单配置

说明

单次请求最多可获取100条清单配置项内容。若需获取超过100条清单配置项，则需发送多次请求，并保留相应的Token，作为下一次请求的参数。

以下代码用于列举某个Bucket的清单配置：

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，并描述脚本用途：列出存储空间清单配置（Inventory）
parser = argparse.ArgumentParser(description="list bucket inventory sample")

# 定义命令行参数，包括必需的区域、存储空间名称、endpoint以及清单ID（可选）
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--inventory_id', help='The name of the inventory.', required=False)  # 清单ID在此场景下是可选的

def main():
    # 解析命令行参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 发送请求以列出指定存储空间的所有清单配置详情
    result = client.list_bucket_inventory(oss.ListBucketInventoryRequest(
            bucket=args.bucket,  # 存储空间名
    ))

    # 打印操作结果的状态码和请求ID，以便确认请求状态
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' list inventory configurations result: {result.list_inventory_configurations_result},'
          f' is truncated: {result.list_inventory_configurations_result.is_truncated},'
          f' next continuation token: {result.list_inventory_configurations_result.next_continuation_token},'
    )

    # 如果存在清单配置，则遍历并打印每个配置的详细信息
    if result.list_inventory_configurations_result.inventory_configurations:
        for r in result.list_inventory_configurations_result.inventory_configurations:
            print(f'result: '
                  f'included object versions: {r.included_object_versions}, '
                  f'optional fields: {r.optional_fields}, '
                  f'id: {r.id}, '
                  f'is enabled: {r.is_enabled}, '
                  f'destination: {r.destination}, '
                  f'schedule: {r.schedule}, '
                  f'filter: {r.filter}'
            )

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始

删除清单配置

以下代码用于删除某个Bucket的清单配置：

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，并描述脚本用途：删除存储空间清单配置（Inventory）
parser = argparse.ArgumentParser(description="delete bucket inventory sample")

# 定义命令行参数，包括必需的区域、存储空间名称、endpoint以及清单ID
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--inventory_id', help='The name of the inventory.', required=True)

def main():
    # 解析命令行参数，获取用户输入的值
    args = parser.parse_args()

    # 从环境变量中加载访问凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK默认配置创建配置对象，并设置认证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置对象的区域属性，根据用户提供的命令行参数
    cfg.region = args.region

    # 如果提供了自定义endpoint，则更新配置对象中的endpoint属性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS客户端，准备与OSS交互
    client = oss.Client(cfg)

    # 发送请求以删除指定存储空间的特定清单配置
    result = client.delete_bucket_inventory(oss.DeleteBucketInventoryRequest(
            bucket=args.bucket,  # 存储空间名
            inventory_id=args.inventory_id,  # 清单配置ID
    ))

    # 打印操作结果的状态码和请求ID，以便确认请求状态
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )

# 当此脚本被直接执行时，调用main函数开始处理逻辑
if __name__ == "__main__":
    main()  # 脚本入口点，控制程序流程从这里开始