如何通过Python SDK V2管理生命周期功能_对象存储(OSS)-阿里云帮助中心

本文介绍如何使用Python SDK V2管理存储空间（Bucket）的生命周期功能。

背景信息

在OSS中，并非所有上传的数据都需要频繁访问，但出于数据合规性或归档需求，部分数据仍需以冷存储的形式保存。根据业务需求，您可以选择：

基于最后修改时间（Last Modified Time）的生命周期规则：当某些数据长时间未被修改且不再需要保留时，可以通过此规则批量删除它们，或将其转换为冷存储类型，从而释放存储空间。
基于最后访问时间（Last Access Time）的生命周期规则：如果希望OSS自动监测数据的访问模式来识别冷数据并动态转换存储类型，可启用此规则。OSS将自动识别长期未被访问的数据，将其转换为更经济的冷存储，从而实现冷热分层，降低存储成本。

注意事项

在配置基于最后一次修改时间或者最后一次访问时间的生命周期规则之前，请确保您已了解该功能。详情请参见基于最后一次修改时间的生命周期规则和基于最后一次访问时间的生命周期规则。
本文示例代码以华东1（杭州）的地域IDcn-hangzhou为例，默认使用外网Endpoint，如果您希望通过与OSS同地域的其他阿里云产品访问OSS，请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系，请参见OSS地域和访问域名。
要设置生命周期规则，您必须有oss:PutBucketLifecycle权限；要查看生命周期规则，您必须有oss:GetBucketLifecycle权限；要清空生命周期规则，您必须有oss:DeleteBucketLifecycle权限。具体操作，请参见为RAM用户授权自定义的权限策略。

设置生命周期规则

以下代码分别提供了设置基于最后一次修改时间和基于最后一次访问时间的生命周期规则的示例。设置完成后，如果您希望修改其中的一条或多条生命周期规则，请参见如何修改其中一条或多条生命周期规则配置？

基于最后一次修改时间策略执行转换文件存储类型

以下代码用于为Bucket设置基于最后一次修改时间策略，以执行转换文件存储类型的操作。

import argparse
import datetime
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，用于接收用户输入的参数
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")

# 添加命令行参数 --region，表示存储空间所在的地域，必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket，表示存储空间的名称，必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint，表示其他服务访问 OSS 时使用的域名，可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint，则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
            bucket=args.bucket,
            lifecycle_configuration=oss.LifecycleConfiguration(
                rules=[oss.LifecycleRule(
                    #  指定生命周期规则rule1。规则中指定前缀为foo、标签键为k1、标签值为v1的文件，在距离最后一次修改时间30天后转为低频访问类型
                    id='rule1',
                    status='Enabled',
                    prefix='foo/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=30,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=False, # 设置为false，基于最后一次修改时间策略
                    )],
                    tags=[oss.Tag(
                        key='k1',
                        value='v1',
                    )],
                ), oss.LifecycleRule(
                    # 指定生命周期规则rule2。规则中指定前缀为dir，在受版本控制状态下的Object仅有删除标记的情况下自动删除删除标记，非当前版本Object超过30天后过期删除，非当前版本Object超过10天后转为IA存储类型
                    id='rule2',
                    status='Enabled',
                    prefix='dir/',
                    expiration=oss.LifecycleRuleExpiration(
                        days=10,
                        expired_object_delete_marker=True
                    ),
                    noncurrent_version_expiration=oss.NoncurrentVersionExpiration(
                        noncurrent_days=30,
                    ),
                    noncurrent_version_transition=oss.NoncurrentVersionTransition(
                        noncurrent_days=10,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=False, # 设置为false，基于最后一次修改时间策略
                    ),
                )]
            ),
    ))

    # 打印操作结果的状态码和请求 ID
    print(f'status code: {result.status_code}, '  # HTTP 状态码，表示请求是否成功
          f'request id: {result.request_id}')    # 请求 ID，用于追踪请求日志和调试


if __name__ == "__main__":
    # 程序入口，调用 main 函数执行逻辑
    main()

基于最后一次修改时间策略限制除指定前缀、标签以外的文件执行转换存储类型操作

以下代码用于指定Bucket中除前缀为log、包含key为key1，value为value1标签且符合指定大小以外的文件在距离最后一次修改时间30天后转低频访问类型。

import argparse
import datetime
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，用于接收用户输入的参数
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")

# 添加命令行参数 --region，表示存储空间所在的地域，必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket，表示存储空间的名称，必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint，表示其他服务访问 OSS 时使用的域名，可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint，则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
            bucket=args.bucket,
            lifecycle_configuration=oss.LifecycleConfiguration(
                rules=[oss.LifecycleRule(
                    #  指定生命周期规则rule1。规则中指定除前缀为log、包含key为key1，value为value1标签且符合指定大小以外的文件在距离最后一次修改时间30天后转低频访问类型
                    id='rule1',
                    status='Enabled',
                    prefix='logs/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=30,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=False, # 设置为false，基于最后一次修改时间策略
                    )],
                    filter=oss.LifecycleRuleFilter(
                        object_size_greater_than=500,
                        object_size_less_than=1000,
                        filter_not=[oss.LifecycleRuleNot(
                            prefix='logs/log',
                            tag={
                               'key': 'key1',
                                'value': 'value1',
                            },
                        )],
                    )
                )]
            ),
    ))

    # 打印操作结果的状态码和请求 ID
    print(f'status code: {result.status_code}, '  # HTTP 状态码，表示请求是否成功
          f'request id: {result.request_id}')    # 请求 ID，用于追踪请求日志和调试


if __name__ == "__main__":
    # 程序入口，调用 main 函数执行逻辑
    main()

基于最后一次访问时间策略转换文件存储类型

以下代码用于为Bucket设置基于最后一次访问时间策略，以执行转换存储类型的操作。

import argparse
import datetime
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，用于接收用户输入的参数
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")

# 添加命令行参数 --region，表示存储空间所在的地域，必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket，表示存储空间的名称，必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint，表示其他服务访问 OSS 时使用的域名，可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint，则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
            bucket=args.bucket,
            lifecycle_configuration=oss.LifecycleConfiguration(
                rules=[oss.LifecycleRule(
                    # 在生命周期规则1中指定前缀为data/的所有文件在距离最后一次访问时间200天后转为低频访问类型。且再次访问前缀为data/的文件时，这些文件仍保留为低频访问类型
                    id='rule1',
                    status='Enabled',
                    prefix='data/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=200,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=True, # 设置为true，基于最后一次访问时间策略
                        return_to_std_when_visit=False
                    )],
                ), oss.LifecycleRule(
                    # 在生命周期规则2中指定前缀为log/的所有文件在距离最后一次访问时间120天后转为低频访问类型。且再次访问前缀为log/的文件时，这些文件仍保留为低频访问类型
		    # 同一规则中指定前缀为log/的所有文件在距离最后一次访问时间250天后转为归档类型。
                    id='rule2',
                    status='Enabled',
                    prefix='log/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=120,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=True, # 设置为true，基于最后一次访问时间策略
                        return_to_std_when_visit=False
                    ), oss.LifecycleRuleTransition(
                        days=250,
                        storage_class=oss.StorageClassType.ARCHIVE,
                        is_access_time=True, # 设置为true，基于最后一次访问时间策略
                        return_to_std_when_visit=False
                    )],
                )]
            ),
    ))

    # 打印操作结果的状态码和请求 ID
    print(f'status code: {result.status_code}, '  # HTTP 状态码，表示请求是否成功
          f'request id: {result.request_id}')    # 请求 ID，用于追踪请求日志和调试


if __name__ == "__main__":
    # 程序入口，调用 main 函数执行逻辑
    main()

查看生命周期规则

以下代码用于查看生命周期规则中包含的信息。

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，用于接收用户输入的参数
parser = argparse.ArgumentParser(description="get bucket lifecycle sample")

# 添加命令行参数 --region，表示存储空间所在的地域，必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket，表示存储空间的名称，必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint，表示其他服务访问 OSS 时使用的域名，可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint，则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    # 调用 get_bucket_lifecycle 方法获取存储空间的生命周期规则
    result = client.get_bucket_lifecycle(
        oss.GetBucketLifecycleRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
        )
    )

    # 打印操作结果的状态码和请求 ID
    print(f'status code: {result.status_code}, '  # HTTP 状态码，表示请求是否成功
          f'request id: {result.request_id}')    # 请求 ID，用于追踪请求日志和调试

    # 如果返回的生命周期规则集合中包含规则，则逐个打印每个规则的详细信息
    if result.lifecycle_configuration.rules:  # 检查是否存在生命周期规则
        for r in result.lifecycle_configuration.rules:  # 遍历所有规则
            print(f'rule: {r}')  # 打印每个规则的内容


if __name__ == "__main__":
    # 程序入口，调用 main 函数执行逻辑
    main()

清空生命周期规则

以下代码用于清空examplebucket的所有生命周期规则。如果您需要删除其中一条或者多条生命周期规则，请参见如何删除其中一条或多条生命周期规则？

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器，用于接收用户输入的参数
parser = argparse.ArgumentParser(description="delete bucket lifecycle sample")

# 添加命令行参数 --region，表示存储空间所在的地域，必填项
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令行参数 --bucket，表示存储空间的名称，必填项
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令行参数 --endpoint，表示其他服务访问 OSS 时使用的域名，可选项
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载 SDK 的默认配置
    cfg = oss.config.load_default()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 设置存储空间所在的地域
    cfg.region = args.region

    # 如果用户提供了自定义的 endpoint，则设置到配置中
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置对象初始化 OSS 客户端
    client = oss.Client(cfg)

    # 调用 delete_bucket_lifecycle 方法删除存储空间的生命周期规则
    result = client.delete_bucket_lifecycle(
        oss.DeleteBucketLifecycleRequest(
            bucket=args.bucket,  # 指定目标存储空间的名称
        )
    )

    # 打印操作结果的状态码和请求 ID
    print(f'status code: {result.status_code}, '  # HTTP 状态码，表示请求是否成功
          f'request id: {result.request_id}')    # 请求 ID，用于追踪请求日志和调试


if __name__ == "__main__":
    # 程序入口，调用 main 函数执行逻辑
    main()

Python生命周期管理