This topic describes how to use Python SDK V2 to manage the lifecycle rules of a bucket.
Background information
In OSS, not all uploaded data requires frequent access. However, for data compliance or data archiving purposes, some data must be saved as cold storage. Based on your business requirements, you can select one of the following options:
Lifecycle rules based on the last modified time: If data has not been modified for a long time and no longer needs to be retained, you can use this type of rule to delete the data in batches or convert its storage class to a cold storage class to free up storage space.
Lifecycle rules based on the last access time: You can enable this type of rule to allow OSS to automatically monitor data access patterns. OSS identifies data that has not been accessed for a long time and converts it to a more cost-effective cold storage class. This process implements Automatic Storage Tiering and reduces storage costs.
Prerequisites
Before you configure a lifecycle rule based on the last modified time or last access time, ensure that you understand the feature. For more information, see Lifecycle rules based on the last modified time and Lifecycle rules based on the last access time.
The sample code in this topic uses the China (Hangzhou) region whose region ID is
cn-hangzhou. By default, a public endpoint 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 the regions and endpoints that are supported by OSS, see OSS regions and endpoints.To set a lifecycle rule, you must have the
oss:PutBucketLifecyclepermission. To view lifecycle rules, you must have theoss:GetBucketLifecyclepermission. To delete all lifecycle rules, you must have theoss:DeleteBucketLifecyclepermission. For more information about how to perform the operations, see Grant custom permissions to a RAM user.
Set lifecycle rules
The following sample code provides examples of how to set lifecycle rules based on the last modified time and last access time. To modify one or more of the lifecycle rules after they are set, see How do I modify the configurations of one or more lifecycle rules?
Convert the storage class of objects based on a policy that uses the last modified time
The following sample code provides an example of how to set a lifecycle rule for a bucket to convert the storage class of objects based on their last modified time.
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# Create a command-line argument parser to receive user-entered parameters.
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument, which specifies the domain name that other services can use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
def main():
# Parse command-line arguments.
args = parser.parse_args()
# Load credentials (AccessKey ID and AccessKey secret) from environment variables.
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# Load the default configurations of the SDK.
cfg = oss.config.load_default()
# Set the credentials provider.
cfg.credentials_provider = credentials_provider
# Set the region where the bucket is located.
cfg.region = args.region
# If a custom endpoint is provided by the user, set it in the configuration.
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# Initialize the OSS client using the configuration object.
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# Specify lifecycle rule rule1. In this rule, objects that have the prefix foo/ and the tag k1:v1 are converted to the Infrequent Access (IA) storage class 30 days after they are last modified.
id='rule1',
status='Enabled',
prefix='foo/',
transitions=[oss.LifecycleRuleTransition(
days=30,
storage_class=oss.StorageClassType.IA,
is_access_time=False, # Set to false, which indicates that the policy is based on the last modified time.
)],
tags=[oss.Tag(
key='k1',
value='v1',
)],
), oss.LifecycleRule(
# Specify lifecycle rule rule2. In this rule, for objects that have the prefix dir/, delete markers are automatically deleted if the objects are in a versioning-enabled bucket and have only delete markers. Non-current versions of objects expire and are deleted after 30 days. Non-current versions of objects are converted to the IA storage class after 10 days.
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, # Set to false, which indicates that the policy is based on the last modified time.
),
)]
),
))
# Print the status code and request ID of the operation.
print(f'status code: {result.status_code}, ' # The HTTP status code, which indicates whether the request is successful.
f'request id: {result.request_id}') # The request ID, which is used to track request logs and for debugging.
if __name__ == "__main__":
# The program entry point that calls the main function to execute the logic.
main()Convert the storage class of objects that do not have a specific prefix or tag based on a policy that uses the last modified time
The following sample code provides an example of how to convert objects in a bucket to the Infrequent Access storage class 30 days after they are last modified. This rule applies to objects that meet the specified size requirements, do not have the prefix log, and do not have a tag with the key key1 and value value1.
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# Create a command-line argument parser to receive user-entered parameters.
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument, which specifies the domain name that other services can use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
def main():
# Parse command-line arguments.
args = parser.parse_args()
# Load credentials (AccessKey ID and AccessKey secret) from environment variables.
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# Load the default configurations of the SDK.
cfg = oss.config.load_default()
# Set the credentials provider.
cfg.credentials_provider = credentials_provider
# Set the region where the bucket is located.
cfg.region = args.region
# If a custom endpoint is provided by the user, set it in the configuration.
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# Initialize the OSS client using the configuration object.
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# Specify lifecycle rule rule1. In this rule, objects that have the prefix logs/ and a size between 500 bytes and 1,000 bytes are converted to the Infrequent Access (IA) storage class 30 days after they are last modified. This rule does not apply to objects that also have the prefix logs/log and the tag key:key1 and value:value1.
id='rule1',
status='Enabled',
prefix='logs/',
transitions=[oss.LifecycleRuleTransition(
days=30,
storage_class=oss.StorageClassType.IA,
is_access_time=False, # Set to false, which indicates that the policy is based on the last modified time.
)],
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',
},
)],
)
)]
),
))
# Print the status code and request ID of the operation.
print(f'status code: {result.status_code}, ' # The HTTP status code, which indicates whether the request is successful.
f'request id: {result.request_id}') # The request ID, which is used to track request logs and for debugging.
if __name__ == "__main__":
# The program entry point that calls the main function to execute the logic.
main()Convert the storage class of objects based on a policy that uses the last access time
The following sample code provides an example of how to set a lifecycle rule for a bucket to convert the storage class of objects based on their last access time.
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# Create a command-line argument parser to receive user-entered parameters.
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument, which specifies the domain name that other services can use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
def main():
# Parse command-line arguments.
args = parser.parse_args()
# Load credentials (AccessKey ID and AccessKey secret) from environment variables.
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# Load the default configurations of the SDK.
cfg = oss.config.load_default()
# Set the credentials provider.
cfg.credentials_provider = credentials_provider
# Set the region where the bucket is located.
cfg.region = args.region
# If a custom endpoint is provided by the user, set it in the configuration.
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# Initialize the OSS client using the configuration object.
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# In lifecycle rule rule1, all objects that have the prefix data/ are converted to the Infrequent Access (IA) storage class 200 days after they are last accessed. When these objects are accessed again, they remain in the IA storage class.
id='rule1',
status='Enabled',
prefix='data/',
transitions=[oss.LifecycleRuleTransition(
days=200,
storage_class=oss.StorageClassType.IA,
is_access_time=True, # Set to true, which indicates that the policy is based on the last access time.
return_to_std_when_visit=False
)],
), oss.LifecycleRule(
# In lifecycle rule rule2, all objects that have the prefix log/ are converted to the Infrequent Access (IA) storage class 120 days after they are last accessed. When these objects are accessed again, they remain in the IA storage class.
# In the same rule, all objects that have the prefix log/ are converted to the Archive storage class 250 days after they are last accessed.
id='rule2',
status='Enabled',
prefix='log/',
transitions=[oss.LifecycleRuleTransition(
days=120,
storage_class=oss.StorageClassType.IA,
is_access_time=True, # Set to true, which indicates that the policy is based on the last access time.
return_to_std_when_visit=False
), oss.LifecycleRuleTransition(
days=250,
storage_class=oss.StorageClassType.ARCHIVE,
is_access_time=True, # Set to true, which indicates that the policy is based on the last access time.
return_to_std_when_visit=False
)],
)]
),
))
# Print the status code and request ID of the operation.
print(f'status code: {result.status_code}, ' # The HTTP status code, which indicates whether the request is successful.
f'request id: {result.request_id}') # The request ID, which is used to track request logs and for debugging.
if __name__ == "__main__":
# The program entry point that calls the main function to execute the logic.
main()View lifecycle rules
The following sample code provides an example of how to view lifecycle rules.
import argparse
import alibabacloud_oss_v2 as oss
# Create a command-line argument parser to receive user-entered parameters.
parser = argparse.ArgumentParser(description="get bucket lifecycle sample")
# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument, which specifies the domain name that other services can use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
def main():
# Parse command-line arguments.
args = parser.parse_args()
# Load credentials (AccessKey ID and AccessKey secret) from environment variables.
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# Load the default configurations of the SDK.
cfg = oss.config.load_default()
# Set the credentials provider.
cfg.credentials_provider = credentials_provider
# Set the region where the bucket is located.
cfg.region = args.region
# If a custom endpoint is provided by the user, set it in the configuration.
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# Initialize the OSS client using the configuration object.
client = oss.Client(cfg)
# Call the get_bucket_lifecycle method to obtain the lifecycle rules of the bucket.
result = client.get_bucket_lifecycle(
oss.GetBucketLifecycleRequest(
bucket=args.bucket, # Specify the name of the destination bucket.
)
)
# Print the status code and request ID of the operation.
print(f'status code: {result.status_code}, ' # The HTTP status code, which indicates whether the request is successful.
f'request id: {result.request_id}') # The request ID, which is used to track request logs and for debugging.
# If the returned collection of lifecycle rules contains rules, print the details of each rule one by one.
if result.lifecycle_configuration.rules: # Check whether lifecycle rules exist.
for r in result.lifecycle_configuration.rules: # Traverse all rules.
print(f'rule: {r}') # Print the content of each rule.
if __name__ == "__main__":
# The program entry point that calls the main function to execute the logic.
main()
Delete all lifecycle rules
The following sample code provides an example of how to delete all lifecycle rules for a bucket named examplebucket. If you want to delete one or more lifecycle rules, see How do I delete one or more lifecycle rules?
import argparse
import alibabacloud_oss_v2 as oss
# Create a command-line argument parser to receive user-entered parameters.
parser = argparse.ArgumentParser(description="delete bucket lifecycle sample")
# Add the --region command-line argument, which specifies the region where the bucket is located. This argument is required.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# Add the --bucket command-line argument, which specifies the name of the bucket. This argument is required.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# Add the --endpoint command-line argument, which specifies the domain name that other services can use to access OSS. This argument is optional.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
def main():
# Parse command-line arguments.
args = parser.parse_args()
# Load credentials (AccessKey ID and AccessKey secret) from environment variables.
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# Load the default configurations of the SDK.
cfg = oss.config.load_default()
# Set the credentials provider.
cfg.credentials_provider = credentials_provider
# Set the region where the bucket is located.
cfg.region = args.region
# If a custom endpoint is provided by the user, set it in the configuration.
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# Initialize the OSS client using the configuration object.
client = oss.Client(cfg)
# Call the delete_bucket_lifecycle method to delete the lifecycle rules of the bucket.
result = client.delete_bucket_lifecycle(
oss.DeleteBucketLifecycleRequest(
bucket=args.bucket, # Specify the name of the destination bucket.
)
)
# Print the status code and request ID of the operation.
print(f'status code: {result.status_code}, ' # The HTTP status code, which indicates whether the request is successful.
f'request id: {result.request_id}') # The request ID, which is used to track request logs and for debugging.
if __name__ == "__main__":
# The program entry point that calls the main function to execute the logic.
main()
References
For the complete sample code for setting a lifecycle rule, see put_bucket_lifecycle.py.
For the complete sample code for viewing lifecycle rules, see get_bucket_lifecycle.py.
For the complete sample code for deleting all lifecycle rules, see delete_bucket_lifecycle.py.