Cara mengelola fitur lifecycle menggunakan Python SDK V2 - Object Storage Service

Aturan lifecycle memungkinkan OSS secara otomatis memindahkan objek ke kelas penyimpanan berbiaya lebih rendah atau menghapusnya berdasarkan usia atau pola akses—tanpa perlu intervensi manual.

OSS mendukung dua jenis aturan:

Aturan berdasarkan waktu modifikasi terakhir: Memindahkan atau menghapus objek yang tidak dimodifikasi selama jumlah hari tertentu. Gunakan aturan ini untuk membersihkan data kedaluwarsa atau mengarsipkan file yang jarang diperbarui.
Aturan berdasarkan waktu akses terakhir: OSS memantau pola akses dan secara otomatis memindahkan data dingin ke kelas penyimpanan yang lebih murah. Fitur ini menerapkan Automatic Storage Tiering dan mengurangi biaya penyimpanan seiring waktu.

Prasyarat

Sebelum memulai, pastikan Anda telah:

Mengenal Aturan lifecycle berdasarkan waktu modifikasi terakhir atau Aturan lifecycle berdasarkan waktu akses terakhir, tergantung pada jenis aturan yang akan Anda konfigurasikan.
Izin berikut diberikan kepada RAM user Anda:
Operation Required permission
Atur aturan lifecycle oss:PutBucketLifecycle
Lihat aturan lifecycle oss:GetBucketLifecycle
Hapus semua aturan lifecycle oss:DeleteBucketLifecycle
Untuk petunjuk pemberian izin, lihat Grant custom permissions to a RAM user.

Operation	Required permission
Atur aturan lifecycle	`oss:PutBucketLifecycle`
Lihat aturan lifecycle	`oss:GetBucketLifecycle`
Hapus semua aturan lifecycle	`oss:DeleteBucketLifecycle`

Kode contoh dalam topik ini menggunakan wilayah China (Hangzhou) (cn-hangzhou) dengan titik akhir publik. Untuk mengakses OSS dari layanan Alibaba Cloud lain di wilayah yang sama, gunakan titik akhir internal sebagai gantinya. Untuk daftar wilayah dan titik akhir yang didukung, lihat OSS regions and endpoints.

Set lifecycle rules

Ketiga contoh di bawah ini memanggil client.put_bucket_lifecycle() untuk menerapkan LifecycleConfiguration ke bucket. Setiap contoh menunjukkan skenario pemfilteran dan transisi yang berbeda.

Untuk memperbarui satu atau beberapa aturan yang telah ada, lihat Bagaimana cara memodifikasi aturan siklus hidup?

Transition objects to IA storage based on last modified time

Contoh ini menetapkan dua aturan pada bucket:

rule1: Memindahkan objek dengan awalan foo/ dan tag k1:v1 ke penyimpanan Infrequent Access (IA) 30 hari setelah modifikasi terakhir.
rule2: Untuk objek dengan awalan dir/ di bucket yang telah diaktifkan versioning—secara otomatis menghapus penanda hapus yang kedaluwarsa, memindahkan versi noncurrent ke IA setelah 10 hari, dan menghapus permanen versi noncurrent setelah 30 hari.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket lifecycle 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()

    # Load credentials from environment variables
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    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.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
        bucket=args.bucket,
        lifecycle_configuration=oss.LifecycleConfiguration(
            rules=[
                oss.LifecycleRule(
                    id='rule1',
                    status='Enabled',
                    prefix='foo/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=30,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=False,  # Policy based on last modified time
                    )],
                    tags=[oss.Tag(
                        key='k1',
                        value='v1',
                    )],
                ),
                oss.LifecycleRule(
                    id='rule2',
                    status='Enabled',
                    prefix='dir/',
                    expiration=oss.LifecycleRuleExpiration(
                        days=10,
                        expired_object_delete_marker=True,  # Remove delete markers when no other versions exist
                    ),
                    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,  # Policy based on last modified time
                    ),
                ),
            ]
        ),
    ))

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


if __name__ == "__main__":
    main()

Filter objects by size and exclude specific prefixes or tags

Contoh ini memindahkan objek dengan awalan logs/ ke penyimpanan Infrequent Access (IA) 30 hari setelah modifikasi terakhir, tetapi hanya jika ukuran objek berada di antara 500 byte dan 1.000 byte. Objek yang juga sesuai dengan awalan logs/log dan memiliki tag key1:value1 dikecualikan.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket lifecycle 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()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    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.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
        bucket=args.bucket,
        lifecycle_configuration=oss.LifecycleConfiguration(
            rules=[
                oss.LifecycleRule(
                    id='rule1',
                    status='Enabled',
                    prefix='logs/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=30,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=False,  # Policy based on last modified time
                    )],
                    filter=oss.LifecycleRuleFilter(
                        object_size_greater_than=500,   # Minimum object size in bytes
                        object_size_less_than=1000,     # Maximum object size in bytes
                        filter_not=[oss.LifecycleRuleNot(
                            prefix='logs/log',          # Exclude this prefix
                            tag={
                                'key': 'key1',
                                'value': 'value1',
                            },
                        )],
                    ),
                ),
            ]
        ),
    ))

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


if __name__ == "__main__":
    main()

Move cold data based on access patterns

Contoh ini menetapkan dua aturan yang memindahkan objek berdasarkan waktu akses terakhir, bukan waktu modifikasi terakhir. Kedua aturan menggunakan is_access_time=True dan return_to_std_when_visit=False.

rule1: Memindahkan objek dengan awalan data/ ke Infrequent Access (IA) setelah 200 hari tidak aktif.
rule2: Memindahkan objek dengan awalan log/ ke IA setelah 120 hari tidak aktif, lalu ke Archive setelah 250 hari.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put bucket lifecycle 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()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    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.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
        bucket=args.bucket,
        lifecycle_configuration=oss.LifecycleConfiguration(
            rules=[
                oss.LifecycleRule(
                    id='rule1',
                    status='Enabled',
                    prefix='data/',
                    transitions=[oss.LifecycleRuleTransition(
                        days=200,
                        storage_class=oss.StorageClassType.IA,
                        is_access_time=True,               # Policy based on last access time
                        return_to_std_when_visit=False,    # Stay in IA when accessed again
                    )],
                ),
                oss.LifecycleRule(
                    id='rule2',
                    status='Enabled',
                    prefix='log/',
                    transitions=[
                        oss.LifecycleRuleTransition(
                            days=120,
                            storage_class=oss.StorageClassType.IA,
                            is_access_time=True,
                            return_to_std_when_visit=False,
                        ),
                        oss.LifecycleRuleTransition(
                            days=250,
                            storage_class=oss.StorageClassType.ARCHIVE,
                            is_access_time=True,
                            return_to_std_when_visit=False,
                        ),
                    ],
                ),
            ]
        ),
    ))

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


if __name__ == "__main__":
    main()

View lifecycle rules

Gunakan client.get_bucket_lifecycle() untuk mengambil semua aturan lifecycle yang dikonfigurasi untuk bucket. Hasilnya berisi daftar lifecycle_configuration.rules yang mencakup definisi lengkap aturan tersebut.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="get bucket lifecycle 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()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    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_lifecycle(
        oss.GetBucketLifecycleRequest(
            bucket=args.bucket,
        )
    )

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

    if result.lifecycle_configuration.rules:
        for r in result.lifecycle_configuration.rules:
            print(f'rule: {r}')


if __name__ == "__main__":
    main()

Delete all lifecycle rules

client.delete_bucket_lifecycle() menghapus semua aturan lifecycle dari bucket. Untuk menghapus aturan tertentu saja, lihat How do I delete specific lifecycle rules?

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="delete bucket lifecycle 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()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    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.delete_bucket_lifecycle(
        oss.DeleteBucketLifecycleRequest(
            bucket=args.bucket,
        )
    )

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


if __name__ == "__main__":
    main()

Referensi

Kode contoh lengkap untuk menetapkan aturan lifecycle: put_bucket_lifecycle.py
Kode contoh lengkap untuk melihat aturan lifecycle: get_bucket_lifecycle.py
Kode contoh lengkap untuk menghapus semua aturan lifecycle: delete_bucket_lifecycle.py