このトピックでは、Python SDK V2 を使用してバケットのライフサイクルルールを管理する方法について説明します。
背景情報
OSS では、アップロードされたすべてのデータに頻繁なアクセスが必要なわけではありません。ただし、データコンプライアンスまたはデータアーカイブの目的で、一部のデータをコールドストレージとして保存する必要があります。ビジネス要件に基づいて、次のオプションのいずれかを選択できます。
最終変更時刻に基づくライフサイクルルール: データが長期間変更されておらず、保持する必要がなくなった場合は、このタイプのルールを使用してデータをバッチで削除するか、ストレージクラスをコールドストレージクラスに変換してストレージスペースを解放できます。
最終アクセス時刻に基づくライフサイクルルール: このタイプのルールを有効にすると、OSS がデータアクセスパターンを自動的に監視できるようになります。 OSS は、長期間アクセスされていないデータを識別し、より費用対効果の高いコールドストレージクラスに変換します。このプロセスは自動ストレージ階層化を実装し、ストレージコストを削減します。
前提条件
最終変更時刻または最終アクセス時刻に基づくライフサイクルルールを構成する前に、この機能を理解していることを確認してください。詳細については、「最終変更時刻に基づくライフサイクルルール」および「最終アクセス時刻に基づくライフサイクルルール」をご参照ください。
このトピックのサンプルコードでは、リージョン ID が
cn-hangzhouの中国 (杭州) リージョンを使用しています。デフォルトでは、パブリックエンドポイントが使用されます。 OSS と同じリージョン内の他の Alibaba Cloud サービスから OSS にアクセスする場合は、内部エンドポイントを使用します。 OSS でサポートされているリージョンとエンドポイントの詳細については、「OSS リージョンとエンドポイント」をご参照ください。ライフサイクルルールを設定するには、
oss:PutBucketLifecycle権限が必要です。ライフサイクルルールを表示するには、oss:GetBucketLifecycle権限が必要です。すべてのライフサイクルルールを削除するには、oss:DeleteBucketLifecycle権限が必要です。操作の実行方法の詳細については、「RAM ユーザーにカスタム権限を付与する」をご参照ください。
ライフサイクルルールを設定する
次のサンプルコードは、最終変更時刻と最終アクセス時刻に基づいてライフサイクルルールを設定する方法の例を示しています。設定後に 1 つ以上のライフサイクルルールを変更するには、「1 つ以上のライフサイクルルールの構成を変更するにはどうすればよいですか?」をご参照ください。
最終変更時刻を使用するポリシーに基づいてオブジェクトのストレージクラスを変換する
次のサンプルコードは、最終変更時刻に基づいてオブジェクトのストレージクラスを変換するバケットのライフサイクルルールを設定する方法の例を示しています。
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# コマンドライン引数パーサーを作成して、ユーザーが入力したパラメーターを受け取ります。
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。この引数は必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。この引数は必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。この引数はオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
def main():
# コマンドライン引数を解析します。
args = parser.parse_args()
# 環境変数から認証情報 (AccessKey ID と AccessKey シークレット) をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# SDK のデフォルト構成をロードします。
cfg = oss.config.load_default()
# 認証情報プロバイダーを設定します。
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを設定します。
cfg.region = args.region
# ユーザーがカスタムエンドポイントを提供している場合は、構成で設定します。
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 構成オブジェクトを使用して OSS クライアントを初期化します。
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# ライフサイクルルール rule1 を指定します。このルールでは、プレフィックス foo/ とタグ k1:v1 を持つオブジェクトは、最終変更から 30 日後に低頻度アクセス (IA) ストレージクラスに変換されます。
id='rule1',
status='Enabled',
prefix='foo/',
transitions=[oss.LifecycleRuleTransition(
days=30,
storage_class=oss.StorageClassType.IA,
is_access_time=False, # false に設定すると、ポリシーが最終変更時刻に基づいていることを示します。
)],
tags=[oss.Tag(
key='k1',
value='v1',
)],
), oss.LifecycleRule(
# ライフサイクルルール rule2 を指定します。このルールでは、プレフィックス dir/ を持つオブジェクトの場合、オブジェクトがバージョン管理対応バケットにあり、削除マーカーのみを持っている場合、削除マーカーは自動的に削除されます。オブジェクトの非カレントバージョンは 30 日後に期限切れになり、削除されます。オブジェクトの非カレントバージョンは 10 日後に IA ストレージクラスに変換されます。
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, # false に設定すると、ポリシーが最終変更時刻に基づいていることを示します。
),
)]
),
))
# 操作のステータスコードとリクエスト ID を出力します。
print(f'status code: {result.status_code}, ' # HTTP ステータスコード。リクエストが成功したかどうかを示します。
f'request id: {result.request_id}') # リクエスト ID。リクエストログの追跡とデバッグに使用されます。
if __name__ == "__main__":
# プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
main()最終変更時刻を使用するポリシーに基づいて、特定のプレフィックスまたはタグを持たないオブジェクトのストレージクラスを変換する
次のサンプルコードは、最終変更から 30 日後にバケット内のオブジェクトを低頻度アクセスストレージクラスに変換する方法の例を示しています。このルールは、指定されたサイズ要件を満たし、プレフィックス log を持たず、キー key1 と値 value1 を持つタグを持たないオブジェクトに適用されます。
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# コマンドライン引数パーサーを作成して、ユーザーが入力したパラメーターを受け取ります。
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。この引数は必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。この引数は必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。この引数はオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
def main():
# コマンドライン引数を解析します。
args = parser.parse_args()
# 環境変数から認証情報 (AccessKey ID と AccessKey シークレット) をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# SDK のデフォルト構成をロードします。
cfg = oss.config.load_default()
# 認証情報プロバイダーを設定します。
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを設定します。
cfg.region = args.region
# ユーザーがカスタムエンドポイントを提供している場合は、構成で設定します。
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 構成オブジェクトを使用して OSS クライアントを初期化します。
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# ライフサイクルルール rule1 を指定します。このルールでは、プレフィックス logs/ を持ち、サイズが 500 バイトから 1,000 バイトの間のオブジェクトは、最終変更から 30 日後に低頻度アクセス (IA) ストレージクラスに変換されます。このルールは、プレフィックス logs/log とタグ key:key1 および value:value1 も持つオブジェクトには適用されません。
id='rule1',
status='Enabled',
prefix='logs/',
transitions=[oss.LifecycleRuleTransition(
days=30,
storage_class=oss.StorageClassType.IA,
is_access_time=False, # false に設定すると、ポリシーが最終変更時刻に基づいていることを示します。
)],
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',
},
)],
)
)]
),
))
# 操作のステータスコードとリクエスト ID を出力します。
print(f'status code: {result.status_code}, ' # HTTP ステータスコード。リクエストが成功したかどうかを示します。
f'request id: {result.request_id}') # リクエスト ID。リクエストログの追跡とデバッグに使用されます。
if __name__ == "__main__":
# プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
main()最終アクセス時刻を使用するポリシーに基づいてオブジェクトのストレージクラスを変換する
次のサンプルコードは、最終アクセス時刻に基づいてオブジェクトのストレージクラスを変換するバケットのライフサイクルルールを設定する方法の例を示しています。
import argparse
import datetime
import alibabacloud_oss_v2 as oss
# コマンドライン引数パーサーを作成して、ユーザーが入力したパラメーターを受け取ります。
parser = argparse.ArgumentParser(description="put bucket lifecycle sample")
# --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。この引数は必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。この引数は必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。この引数はオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
def main():
# コマンドライン引数を解析します。
args = parser.parse_args()
# 環境変数から認証情報 (AccessKey ID と AccessKey シークレット) をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# SDK のデフォルト構成をロードします。
cfg = oss.config.load_default()
# 認証情報プロバイダーを設定します。
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを設定します。
cfg.region = args.region
# ユーザーがカスタムエンドポイントを提供している場合は、構成で設定します。
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 構成オブジェクトを使用して OSS クライアントを初期化します。
client = oss.Client(cfg)
result = client.put_bucket_lifecycle(oss.PutBucketLifecycleRequest(
bucket=args.bucket,
lifecycle_configuration=oss.LifecycleConfiguration(
rules=[oss.LifecycleRule(
# ライフサイクルルール rule1 では、プレフィックス data/ を持つすべてのオブジェクトは、最終アクセスから 200 日後に低頻度アクセス (IA) ストレージクラスに変換されます。これらのオブジェクトに再びアクセスしても、IA ストレージクラスのままです。
id='rule1',
status='Enabled',
prefix='data/',
transitions=[oss.LifecycleRuleTransition(
days=200,
storage_class=oss.StorageClassType.IA,
is_access_time=True, # true に設定すると、ポリシーが最終アクセス時刻に基づいていることを示します。
return_to_std_when_visit=False
)],
), oss.LifecycleRule(
# ライフサイクルルール rule2 では、プレフィックス log/ を持つすべてのオブジェクトは、最終アクセスから 120 日後に低頻度アクセス (IA) ストレージクラスに変換されます。これらのオブジェクトに再びアクセスしても、IA ストレージクラスのままです。
# 同じルールでは、プレフィックス log/ を持つすべてのオブジェクトは、最終アクセスから 250 日後にアーカイブストレージクラスに変換されます。
id='rule2',
status='Enabled',
prefix='log/',
transitions=[oss.LifecycleRuleTransition(
days=120,
storage_class=oss.StorageClassType.IA,
is_access_time=True, # true に設定すると、ポリシーが最終アクセス時刻に基づいていることを示します。
return_to_std_when_visit=False
), oss.LifecycleRuleTransition(
days=250,
storage_class=oss.StorageClassType.ARCHIVE,
is_access_time=True, # true に設定すると、ポリシーが最終アクセス時刻に基づいていることを示します。
return_to_std_when_visit=False
)],
)]
),
))
# 操作のステータスコードとリクエスト ID を出力します。
print(f'status code: {result.status_code}, ' # HTTP ステータスコード。リクエストが成功したかどうかを示します。
f'request id: {result.request_id}') # リクエスト ID。リクエストログの追跡とデバッグに使用されます。
if __name__ == "__main__":
# プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
main()ライフサイクルルールを表示する
次のサンプルコードは、ライフサイクルルールを表示する方法の例を示しています。
import argparse
import alibabacloud_oss_v2 as oss
# コマンドライン引数パーサーを作成して、ユーザーが入力したパラメーターを受け取ります。
parser = argparse.ArgumentParser(description="get bucket lifecycle sample")
# --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。この引数は必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。この引数は必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。この引数はオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
def main():
# コマンドライン引数を解析します。
args = parser.parse_args()
# 環境変数から認証情報 (AccessKey ID と AccessKey シークレット) をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# SDK のデフォルト構成をロードします。
cfg = oss.config.load_default()
# 認証情報プロバイダーを設定します。
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを設定します。
cfg.region = args.region
# ユーザーがカスタムエンドポイントを提供している場合は、構成で設定します。
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 構成オブジェクトを使用して OSS クライアントを初期化します。
client = oss.Client(cfg)
# get_bucket_lifecycle メソッドを呼び出して、バケットのライフサイクルルールを取得します。
result = client.get_bucket_lifecycle(
oss.GetBucketLifecycleRequest(
bucket=args.bucket, # デスティネーションバケットの名前を指定します。
)
)
# 操作のステータスコードとリクエスト ID を出力します。
print(f'status code: {result.status_code}, ' # HTTP ステータスコード。リクエストが成功したかどうかを示します。
f'request id: {result.request_id}') # リクエスト ID。リクエストログの追跡とデバッグに使用されます。
# 返されたライフサイクルルールのコレクションにルールが含まれている場合は、各ルールの詳細を 1 つずつ出力します。
if result.lifecycle_configuration.rules: # ライフサイクルルールが存在するかどうかを確認します。
for r in result.lifecycle_configuration.rules: # すべてのルールをトラバースします。
print(f'rule: {r}') # 各ルールの内容を出力します。
if __name__ == "__main__":
# プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
main()
すべてのライフサイクルルールを削除する
次のサンプルコードは、examplebucket という名前のバケットのすべてのライフサイクルルールを削除する方法の例を示しています。 1 つ以上のライフサイクルルールを削除する場合は、「1 つ以上のライフサイクルルールを削除するにはどうすればよいですか?」をご参照ください。
import argparse
import alibabacloud_oss_v2 as oss
# コマンドライン引数パーサーを作成して、ユーザーが入力したパラメーターを受け取ります。
parser = argparse.ArgumentParser(description="delete bucket lifecycle sample")
# --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。この引数は必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。この引数は必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。この引数はオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
def main():
# コマンドライン引数を解析します。
args = parser.parse_args()
# 環境変数から認証情報 (AccessKey ID と AccessKey シークレット) をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# SDK のデフォルト構成をロードします。
cfg = oss.config.load_default()
# 認証情報プロバイダーを設定します。
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを設定します。
cfg.region = args.region
# ユーザーがカスタムエンドポイントを提供している場合は、構成で設定します。
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 構成オブジェクトを使用して OSS クライアントを初期化します。
client = oss.Client(cfg)
# delete_bucket_lifecycle メソッドを呼び出して、バケットのライフサイクルルールを削除します。
result = client.delete_bucket_lifecycle(
oss.DeleteBucketLifecycleRequest(
bucket=args.bucket, # デスティネーションバケットの名前を指定します。
)
)
# 操作のステータスコードとリクエスト ID を出力します。
print(f'status code: {result.status_code}, ' # HTTP ステータスコード。リクエストが成功したかどうかを示します。
f'request id: {result.request_id}') # リクエスト ID。リクエストログの追跡とデバッグに使用されます。
if __name__ == "__main__":
# プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
main()
参考文献
ライフサイクルルールを設定するための完全なサンプルコードについては、put_bucket_lifecycle.py を参照してください。
ライフサイクルルールを表示するための完全なサンプルコードについては、get_bucket_lifecycle.py を参照してください。
すべてのライフサイクルルールを削除するための完全なサンプルコードについては、delete_bucket_lifecycle.py を参照してください。