Use the Python SDK to reduce storage costs with lifecycle rules - Object Storage Service

Not all data uploaded to Object Storage Service (OSS) requires frequent access. Some data must be kept in cold storage for compliance or archival purposes. You can configure lifecycle rules based on the last modified time to batch delete data that is no longer needed in a bucket. You can also configure lifecycle rules based on the last access time to achieve automatic storage tiering and reduce storage costs. OSS then automatically monitors data access patterns, identifies cold data, and transitions the data to a different storage class.

Usage notes

Before you configure lifecycle rules based on the last modified time or last access time of objects, make sure that you familiarize yourself with this feature. For more information, see Lifecycle rules based on the last modified time and Lifecycle rules based on the last access time.

In this topic, the public endpoint of the China (Hangzhou) region is used. If you want to access OSS from other Alibaba Cloud services in the same region as OSS, use an internal endpoint. For more information about OSS regions and endpoints, see Regions and endpoints.
In this topic, access credentials are obtained from environment variables. For more information about how to configure access credentials, see Configure access credentials using OSS SDK for Python 1.0.
In this topic, an OSSClient instance is created by using an OSS endpoint. If you want to create an OSSClient instance by using custom domain names or Security Token Service (STS), see Initialization.
To configure lifecycle rules, you must have the oss:PutBucketLifecycle permission. To query lifecycle rules, you must have the oss:GetBucketLifecycle permission. To clear lifecycle rules, you must have the oss:DeleteBucketLifecycle permission. For more information, see Attach a custom policy to a RAM user.

Set lifecycle rules

The following code provides examples of how to set lifecycle rules based on the last modified time and the last access time. For information about how to modify one or more rules after they are set, see How do I modify one or more lifecycle rule configurations?.

Transition storage classes and delete objects based on the last modified time

The following code provides an example of how to set a lifecycle rule for the `examplebucket` bucket. This rule transitions the storage class of matched objects and deletes them based on their last modified time.

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
import datetime
from oss2.models import (LifecycleExpiration, LifecycleRule,
                        BucketLifecycle, AbortMultipartUpload,
                        TaggingRule, Tagging, StorageTransition,
                        NoncurrentVersionStorageTransition,
                        NoncurrentVersionExpiration)

# Obtain access credentials from environment variables. Before you run this sample code, make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint of the region where the bucket is located. For example, if the bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region information that corresponds to the endpoint, for example, cn-hangzhou. Note that this parameter is required for V4 signatures.
region = "cn-hangzhou"

# Replace examplebucket with the actual bucket name.
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

# Set objects to expire 3 days after their last modified time.
rule1 = LifecycleRule('rule1', 'tests/',
                      status=LifecycleRule.ENABLED,
                      expiration=LifecycleExpiration(days=3))

# Set an object expiration rule. Objects created before the specified date expire.
rule2 = LifecycleRule('rule2', 'tests2/',
                      status=LifecycleRule.ENABLED,
expiration = LifecycleExpiration(created_before_date=datetime.date(2023, 12, 12)))

# Set a part expiration rule. Parts expire after 3 days.
rule3 = LifecycleRule('rule3', 'tests3/',
                      status=LifecycleRule.ENABLED,
            abort_multipart_upload=AbortMultipartUpload(days=3))

# Set a part expiration rule. Parts created before the specified date expire.
rule4 = LifecycleRule('rule4', 'tests4/',
                      status=LifecycleRule.ENABLED,
                      abort_multipart_upload = AbortMultipartUpload(created_before_date=datetime.date(2022, 12, 12)))

# Set a storage class transition rule. Transition objects to Infrequent Access (IA) 20 days after their last modified time. Transition them to Archive 30 days after their last modified time.
rule5 = LifecycleRule('rule5', 'tests5/',
                      status=LifecycleRule.ENABLED,
                      storage_transitions=[StorageTransition(days=20,storage_class=oss2.BUCKET_STORAGE_CLASS_IA),
                            StorageTransition(days=30,storage_class=oss2.BUCKET_STORAGE_CLASS_ARCHIVE)])

# Set matching tags.
tagging_rule = TaggingRule()
tagging_rule.add('key1', 'value1')
tagging_rule.add('key2', 'value2')
tagging = Tagging(tagging_rule)

# Set a storage class transition rule. Transition objects to the Archive storage class 365 days after their last modified time. 
# Unlike the preceding rules, rule6 specifies matching tags. Only objects that have both the key1=value1 and key2=value2 tags match this rule.
rule6 = LifecycleRule('rule6', 'tests6/',
                      status=LifecycleRule.ENABLED,
                      storage_transitions=[StorageTransition(created_before_date=datetime.date(2022, 12, 12),storage_class=oss2.BUCKET_STORAGE_CLASS_IA)],
                      tagging = tagging)

# rule7 applies to a bucket with versioning enabled.
# Automatically transition objects to the Archive storage class 365 days after their last modified time.
# Automatically remove expired delete markers.
# Transition noncurrent versions of objects to the IA storage class after 12 days.
# Transition noncurrent versions of objects to the Archive storage class after 20 days.
# Delete noncurrent versions of objects after 30 days.
rule7 = LifecycleRule('rule7', 'tests7/',
              status=LifecycleRule.ENABLED,
              storage_transitions=[StorageTransition(days=365, storage_class=oss2.BUCKET_STORAGE_CLASS_ARCHIVE)],
              expiration=LifecycleExpiration(expired_detete_marker=True),
              noncurrent_version_sotrage_transitions =
                    [NoncurrentVersionStorageTransition(12, oss2.BUCKET_STORAGE_CLASS_IA),
                     NoncurrentVersionStorageTransition(20, oss2.BUCKET_STORAGE_CLASS_ARCHIVE)],
              noncurrent_version_expiration = NoncurrentVersionExpiration(30))

lifecycle = BucketLifecycle([rule1, rule2, rule3, rule4, rule5, rule6, rule7])

bucket.put_bucket_lifecycle(lifecycle)

Transition the storage class of objects that do not have a specific prefix and tag based on the last modified time

The following code provides an example of how to use the Not element in the filter node. In this example, for the `examplebucket` bucket, all files except those with the `logs/not-prefix` prefix and the `key1:value1` tag are transitioned to the Infrequent Access storage class 30 days after their last modified time.

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from oss2.models import LifecycleRule, BucketLifecycle, StorageTransition, LifecycleFilter, FilterNot, FilterNotTag

# Obtain access credentials from environment variables. Before you run this sample code, make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint of the region where the bucket is located. For example, if the bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region information that corresponds to the endpoint, for example, cn-hangzhou. Note that this parameter is required for V4 signatures.
region = "cn-hangzhou"

# Replace examplebucket with the actual bucket name.
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

not_prefix = 'logs/not-prefix'
key = 'key1'
value = 'value1'
not_tag = FilterNotTag(key, value)
filter_not = FilterNot(not_prefix, not_tag)
filter = LifecycleFilter([filter_not])

# Set the storage class transition rule.
rule1 = LifecycleRule('mtime transition1', 'logs',
                      status=LifecycleRule.ENABLED,
                      storage_transitions=[StorageTransition(days=30, storage_class=oss2.BUCKET_STORAGE_CLASS_IA)],
                      filter=filter)

lifecycle = BucketLifecycle([rule1])

result = bucket.put_bucket_lifecycle(lifecycle)
print('Lifecycle rule set successfully. Status code: ' + str(result.status))

Transition storage classes based on the last access time

The following code provides an example of how to set a lifecycle rule for the `examplebucket` bucket to transition storage classes based on the last access time.

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from oss2.models import LifecycleRule, BucketLifecycle, StorageTransition, NoncurrentVersionStorageTransition

# Obtain access credentials from environment variables. Before you run this sample code, make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint of the region where the bucket is located. For example, if the bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region information that corresponds to the endpoint, for example, cn-hangzhou. Note that this parameter is required for V4 signatures.
region = "cn-hangzhou"

# Replace examplebucket with the actual bucket name.
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

# Specify lifecycle rule 1. The rule transitions all files that have the prefix 'logs' and are larger than 64 KB to the Infrequent Access storage class 30 days after their last access time. When these files are accessed again, they remain in the Infrequent Access storage class.
rule1 = LifecycleRule('rule1', 'logs', status=LifecycleRule.ENABLED)
rule1.storage_transitions = [StorageTransition(days=30, storage_class=oss2.BUCKET_STORAGE_CLASS_IA, is_access_time=True, return_to_std_when_visit=False, allow_small_file=True)]

# Specify lifecycle rule 2. The rule transitions all noncurrent versions of files that have the prefix 'dir' and are larger than 64 KB to the Infrequent Access storage class 10 days after their last access time. When these files are accessed again, they are transitioned to the Standard storage class.
rule2 = LifecycleRule('rule2', 'dir', status=LifecycleRule.ENABLED)
rule2.noncurrent_version_sotrage_transitions = [NoncurrentVersionStorageTransition(10, oss2.BUCKET_STORAGE_CLASS_IA, is_access_time=True, return_to_std_when_visit=True, allow_small_file=False)]

lifecycle = BucketLifecycle([rule1, rule2])

# Set the lifecycle rule.
result = bucket.put_bucket_lifecycle(lifecycle)

print('Lifecycle rule set successfully. Status code: ' + str(result.status))

View lifecycle rules

The following code provides an example of how to view the lifecycle rules for the `examplebucket` bucket.

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# Obtain access credentials from environment variables. Before you run this sample code, make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint of the region where the bucket is located. For example, if the bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region information that corresponds to the endpoint, for example, cn-hangzhou. Note that this parameter is required for V4 signatures.
region = "cn-hangzhou"

# Replace examplebucket with the actual bucket name.
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

# View the lifecycle rules.
lifecycle = bucket.get_bucket_lifecycle()

for rule in lifecycle.rules:
    print('==========')
    # View the rule ID.
    print('id:', rule.id)
    # View the rule prefix.
    print('prefix:', rule.prefix)
    # View the rule status.
    print('status', rule.status)
    if rule.tagging is not None:
        # View the rule tags.
        print('tagging:', rule.tagging)

    if rule.abort_multipart_upload is not None:
        if rule.abort_multipart_upload.days is not None:
            # View the part expiration rule.
            print('abort_multipart_upload days:', rule.abort_multipart_upload.days)
        else:
            print('abort_multipart_upload created_before_date:', rule.abort_multipart_upload.created_before_date)

    if rule.expiration is not None:
        if rule.expiration.days is not None:
            # View the number of days until expiration.
            print('expiration days:', rule.expiration.days)
        elif rule.expiration.expired_detete_marker is not None:
            # Check whether expired delete markers are automatically removed.
            print('expiration expired_detete_marker:', rule.expiration.expired_detete_marker)
        elif rule.expiration.created_before_date is not None:
            # View the expiration date.
            print('expiration created_before_date:', rule.expiration.created_before_date)

    if len(rule.storage_transitions) > 0:
        storage_info = ''
        for storage_rule in rule.storage_transitions:
            if storage_rule.days is not None:
                storage_info += 'days={0}, storage_class={1} *** '.format(
                    storage_rule.days, storage_rule.storage_class)
            else:
                storage_info += 'created_before_date={0}, storage_class={1} *** '.format(
                    storage_rule.created_before_date, storage_rule.storage_class)
            # Check whether the lifecycle rule is configured based on the last access time. This check is supported only in Python SDK 2.16.1 and later.
            storage_info += 'is_access_time={0}, return_to_std_when_visit={1} ***'.format(
                storage_rule.is_access_time, storage_rule.return_to_std_when_visit)
        print('storage_transitions:', storage_info)

    if len(rule.noncurrent_version_sotrage_transitions) > 0:
        noncurrent_storage_info = ''
        for storage_rule in rule.noncurrent_version_sotrage_transitions:
            noncurrent_storage_info += 'days={0}, storage_class={1} *** '.format(
                storage_rule.noncurrent_days, storage_rule.storage_class)
            # Check whether an object is transitioned to the Standard storage class when it is accessed again after being transitioned to the Infrequent Access storage class. This check is supported only in Python SDK 2.16.1 and later.
            noncurrent_storage_info += 'is_access_time={0}, return_to_std_when_visit={1} ***'.format(
                storage_rule.is_access_time, storage_rule.return_to_std_when_visit)
        # View the storage class transition rule for noncurrent versions of objects.
        print('noncurrent_version_sotrage_transitions:', noncurrent_storage_info)

    if rule.noncurrent_version_expiration is not None:
        # View the expiration rule for noncurrent versions of objects.
        print('noncurrent_version_expiration days:', rule.noncurrent_version_expiration.noncurrent_days)

Delete lifecycle rules

The following code provides an example of how to delete all lifecycle rules for the `examplebucket` bucket. For information about how to delete one or more specific lifecycle rules, see How do I delete one or more lifecycle rules?.

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# Obtain access credentials from environment variables. Before you run this sample code, make sure that the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables are set.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint of the region where the bucket is located. For example, if the bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region information that corresponds to the endpoint, for example, cn-hangzhou. Note that this parameter is required for V4 signatures.
region = "cn-hangzhou"

# Replace examplebucket with the actual bucket name.
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

# Delete the lifecycle rules.
bucket.delete_bucket_lifecycle()

# Viewing the lifecycle rules again throws an exception.
try:
    lifecycle = bucket.get_bucket_lifecycle()
except oss2.exceptions.NoSuchLifecycle:
    print('lifecycle is not configured')

References

For the complete sample code for lifecycle rules, see GitHub examples.
For more information about the API operation used to set lifecycle rules, see PutBucketLifecycle.
For more information about the API operation used to view lifecycle rules, see GetBucketLifecycle.
For more information about the API operation used to delete lifecycle rules, see DeleteBucketLifecycle.