如何使用Python SDK V2管理对象级别保留策略（Object Worm）-对象存储(OSS)-阿里云帮助中心

对象级别保留策略（Object Worm）提供对象粒度的WORM合规保留能力，支持为每个Object设置独立的保留模式和保留截止日期。本文介绍如何使用Python SDK V2开启Object Worm、查询配置、设置和查询对象保留策略。

注意事项

重要

该功能目前为邀测功能，如需使用，请联系技术支持申请开通。当前仅支持华东1（杭州）、西南1（成都）、华南1（深圳）、华南3（广州）、华北1（青岛）、华北2（北京）、华北5（呼和浩特）、华南2（河源）、中国香港、德国（法兰克福）、日本（东京）、韩国（首尔）、新加坡、马来西亚（吉隆坡）、印度尼西亚（雅加达）、菲律宾（马尼拉）、泰国（曼谷）、沙特（利雅得）、阿联酋（迪拜）地域。

本文示例代码以华东1（杭州）的地域IDcn-hangzhou为例，默认使用外网Endpoint，如果您希望通过与OSS同地域的其他阿里云产品访问OSS，请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系，请参见地域和Endpoint。

示例代码

开启Object Worm并配置默认保留策略

您可以使用以下代码开启Object Worm并配置默认保留策略：

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket object worm configuration sample")
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('--mode', help='Object-level retention strategy pattern. Valid values: GOVERNANCE, COMPLIANCE', default='GOVERNANCE')
parser.add_argument('--days', help='Object-level retention policy days (max 36500)', type=int)
parser.add_argument('--years', help='Bucket object level retention policy years (max 100)', type=int)


def main():

    args = parser.parse_args()

    # Loading credentials values from the environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Using the SDK's default configuration
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Create default retention settings
    default_retention = oss.ObjectWormConfigurationRuleDefaultRetention(
        mode=args.mode,
        days=args.days,
        # days and years can only appear once
        # years=args.years,
    )

    # Create rule container
    rule = oss.ObjectWormConfigurationRule(
        default_retention=default_retention,
    )

    # Create configuration
    config = oss.ObjectWormConfiguration(
        object_worm_enabled='Enabled',
        rule=rule,
    )

    result = client.put_bucket_object_worm_configuration(oss.PutBucketObjectWormConfigurationRequest(
        bucket=args.bucket,
        object_worm_configuration=config,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
    )


if __name__ == "__main__":
    main()

查询Object Worm配置

您可以使用以下代码查询Bucket的Object Worm配置：

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="get bucket object worm configuration sample")
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')


def main():
    args = parser.parse_args()

    # Loading credentials values from the environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Using the SDK's default configuration
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    result = client.get_bucket_object_worm_configuration(oss.GetBucketObjectWormConfigurationRequest(
        bucket=args.bucket,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          )

    if result.object_worm_configuration:
        worm_config = result.object_worm_configuration
        print(f'object worm enabled: {worm_config.object_worm_enabled}')

        if worm_config.rule and worm_config.rule.default_retention:
            retention = worm_config.rule.default_retention
            print(f'retention mode: {retention.mode}')
            print(f'retention days: {retention.days}')
            print(f'retention years: {retention.years}')


if __name__ == "__main__":
    main()

设置对象保留策略

您可以使用以下代码为Object设置独立的保留策略：

import argparse
import alibabacloud_oss_v2 as oss
from datetime import datetime, timedelta, timezone

parser = argparse.ArgumentParser(description="put object retention sample")
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('--key', help='The name of the object.', required=True)

def main():

    args = parser.parse_args()

    # Loading credentials values from the environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Using the SDK's default configuration
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Calculate retain until date (1 days from now) in ISO 8601 format
    # Use UTC time (recommended for OSS)
    retain_until_date = datetime.now(timezone.utc) + timedelta(days=1)
    retain_until_iso = retain_until_date.strftime('%Y-%m-%dT%H:%M:%S.000Z')


    # Create retention configuration
    retention_config = oss.Retention(
        mode=oss.ObjectRetentionModeType.COMPLIANCE,
        retain_until_date=retain_until_iso,
    )

    # Set object retention
    result = client.put_object_retention(oss.PutObjectRetentionRequest(
        bucket=args.bucket,
        key=args.key,
        retention=retention_config,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}')


if __name__ == "__main__":
    main()

查询对象保留策略

您可以使用以下代码查询Object的保留策略：

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="get object retention sample")
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('--key', help='The name of the object.', required=True)


def main():

    args = parser.parse_args()

    # Loading credentials values from the environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Using the SDK's default configuration
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Get object retention configuration
    result = client.get_object_retention(oss.GetObjectRetentionRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' retention mode: {result.retention.mode},'
          f' retain until date: {result.retention.retain_until_date}')


if __name__ == "__main__":
    main()