How to manage Object Worm with the Python SDK V2-Object Storage Service(OSS)-阿里云帮助中心

Object Worm provides object-level WORM (Write Once, Read Many) compliance, allowing you to set a unique retention mode and retention expiration date for each object. You can use the Python SDK V2 to enable Object Worm, query its configuration, and set and query object retention policies.

Usage notes

Important

This feature is currently available by invitation only. To request access, contact technical support.

The code examples in this topic use the China (Hangzhou) region (region ID: cn-hangzhou) as an example. By default, the examples use a public endpoint. To access OSS from other Alibaba Cloud services within the same region, use an internal endpoint instead. For a complete list of OSS endpoints by region, see Regions and Endpoints.

Code examples

Configure Object Worm

The following code enables Object Worm for a bucket and configures a default retention policy:

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 OSS endpoint. If not specified, the SDK determines the endpoint based on the region.')
parser.add_argument('--mode', help='The retention mode. 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 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 a default retention rule
    default_retention = oss.ObjectWormConfigurationRuleDefaultRetention(
        mode=args.mode,
        days=args.days,
        # Specify either days or years, but not both.
        # years=args.years,
    )

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

    # Create the WORM 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()

Query Object Worm configuration

The following code queries the Object Worm configuration of a bucket:

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 OSS endpoint. If not specified, the SDK determines the endpoint based on the region.')


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

    # Loading credentials 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()

Set object retention policy

The following code sets a retention policy for an 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 OSS endpoint. If not specified, the SDK determines the endpoint based on the region.')
parser.add_argument('--key', help='The object key.', required=True)

def main():

    args = parser.parse_args()

    # Loading credentials 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 day 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()

Query object retention policy

The following code queries the retention policy of an 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 OSS endpoint. If not specified, the SDK determines the endpoint based on the region.')
parser.add_argument('--key', help='The object key.', required=True)


def main():

    args = parser.parse_args()

    # Loading credentials 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()