Object Storage Service：Replikasi data menggunakan OSS SDK untuk Python 2.0

Contoh kode berikut mengaktifkan replikasi data dan membuat tugas replikasi data yang mereplikasi data dari bucket sumber ke bucket tujuan di wilayah yang sama atau berbeda:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="put bucket replication sample")
# Tentukan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
parser.add_argument('--sync_role', help='Peran yang ingin Anda otorisasi OSS untuk digunakan untuk mereplikasi data', required=True)
parser.add_argument('--target_bucket', help='Bucket tujuan ke mana data direplikasi', required=True)
parser.add_argument('--target_location', help='Wilayah tempat bucket tujuan berada', required=True)

def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan untuk autentikasi.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()
    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider
    # Tentukan wilayah.
    cfg.region = args.region
    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Konfigurasikan dan jalankan permintaan PutBucketReplication.
    result = client.put_bucket_replication(oss.PutBucketReplicationRequest(
            bucket=args.bucket,  # Nama bucket sumber.
            replication_configuration=oss.ReplicationConfiguration(
                rules=[oss.ReplicationRule(
                    source_selection_criteria=oss.ReplicationSourceSelectionCriteria(
                        sse_kms_encrypted_objects=oss.SseKmsEncryptedObjects(
                            status=oss.StatusType.ENABLED,
                        ),
                    ),
                    rtc=oss.ReplicationTimeControl(
                        status='disabled',  # Nonaktifkan RTC.
                    ),
                    destination=oss.ReplicationDestination(
                        bucket=args.target_bucket,  # Nama bucket tujuan.
                        location=args.target_location,  # Wilayah bucket tujuan.
                        transfer_type=oss.TransferType.INTERNAL,  # Jenis transfer.
                    ),
                    historical_object_replication=oss.HistoricalObjectReplicationType.DISABLED,  # Larang replikasi data historis.
                    sync_role=args.sync_role,  # Peran untuk replikasi data.
                    status='Disabled',  # Nonaktifkan aturan.
                    prefix_set=oss.ReplicationPrefixSet(
                        prefixs=['aaa/', 'bbb/'],  # Awalan dalam nama objek yang akan direplikasi.
                    ),
                    action='ALL',  # Semua operasi.
                )],
            ),
    ))

    # Tampilkan kode status HTTP dan ID permintaan.
    print(f'kode status: {result.status_code}, ID permintaan: {result.request_id}')

if __name__ == "__main__":
    main()

Kode berikut menanyakan aturan replikasi data untuk bucket:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="get bucket replication sample")
# Tambahkan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')

def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()
    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider
    # Tentukan wilayah.
    cfg.region = args.region
    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Tanyakan aturan replikasi untuk bucket.
    result = client.get_bucket_replication(oss.GetBucketReplicationRequest(
            bucket=args.bucket,
    ))

    # Tampilkan informasi respons dasar.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},'
          f' konfigurasi replikasi: {result.replication_configuration}')

    # Jika ada aturan replikasi data, tampilkan detail aturan.
    if result.replication_configuration.rules:
        for r in result.replication_configuration.rules:
            print(f'hasil: kriteria pemilihan sumber: {r.source_selection_criteria}, '
                  f'rtc: {r.rtc}, tujuan: {r.destination}, '
                  f'replikasi objek historis: {r.historical_object_replication}, '
                  f'peran sinkronisasi: {r.sync_role}, status: {r.status}, '
                  f'konfigurasi enkripsi: {r.encryption_configuration}, '
                  f'id: {r.id}, set awalan: {r.prefix_set}, aksi: {r.action}')

if __name__ == "__main__":
    main()

Contoh kode berikut mengaktifkan atau menonaktifkan fitur RTC untuk aturan CRR:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="put bucket rtc sample")

# Tambahkan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
parser.add_argument('--status', help='Tentukan apakah akan mengaktifkan RTC. Nilai valid: disabled, enabled', default='disabled')
parser.add_argument('--rule_id', help='ID aturan replikasi data untuk mana Anda ingin mengonfigurasi RTC.', required=True)

def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()
    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider
    # Tentukan wilayah.
    cfg.region = args.region
    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Buat konfigurasi status RTS dan terapkan pada bucket tertentu.
    result = client.put_bucket_rtc(oss.PutBucketRtcRequest(
            bucket=args.bucket,
            rtc_configuration=oss.RtcConfiguration(
                rtc=oss.ReplicationTimeControl(
                    status=args.status,
                ),
                id=args.rule_id,
            ),
    ))

    # Tampilkan informasi respons dasar.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id}')

if __name__ == "__main__":
    main()

Contoh kode berikut menanyakan wilayah ke mana data dapat direplikasi dari bucket sumber yang ditentukan:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="get bucket replication location sample")
# Tambahkan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')


def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()
    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider
    # Tentukan wilayah.
    cfg.region = args.region
    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Tanyakan wilayah ke mana data di bucket yang ditentukan dapat direplikasi.
    result = client.get_bucket_replication_location(oss.GetBucketReplicationLocationRequest(
            bucket=args.bucket,
    ))

    # Tampilkan informasi respons dasar.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},'
          f' lokasi replikasi: {result.replication_location},'
          f' batasan jenis transfer lokasi: {result.replication_location.location_transfer_type_constraint},'
          f' batasan lokasirtc: {result.replication_location.locationrtc_constraint},'
    )

    # Jika ada batasan pada wilayah tujuan untuk replikasi data, tampilkan detail wilayah dan jenis transfer.
    if result.replication_location.location_transfer_type_constraint.location_transfer_types:
        for r in result.replication_location.location_transfer_type_constraint.location_transfer_types:
            print(f'hasil: lokasi: {r.location}, jenis transfer: {r.transfer_types}')

if __name__ == "__main__":
    main()

Contoh kode berikut menanyakan kemajuan tugas replikasi data yang memiliki ID aturan replikasi tertentu untuk bucket:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="get bucket replication progress sample")
# Tambahkan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
parser.add_argument('--rule_id', help='ID aturan replikasi data untuk mana Anda ingin mengonfigurasi RTC.', required=True)

def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()
    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider
    # Tentukan wilayah.
    cfg.region = args.region
    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Tanyakan kemajuan replikasi data.
    result = client.get_bucket_replication_progress(oss.GetBucketReplicationProgressRequest(
            bucket=args.bucket,
            rule_id=args.rule_id,
    ))

    # Tampilkan informasi respons dasar.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},'
          f' kemajuan replikasi: {result.replication_progress}')

    # Jika aturan replikasi data ada, tampilkan detail aturan.
    if result.replication_progress.rules:
        for r in result.replication_progress.rules:
            print(f'hasil: replikasi objek historis: {r.historical_object_replication}, '
                  f'kemajuan: {r.progress}, '
                  f'id: {r.id}, '
                  f'set awalan: {r.prefix_set}, '
                  f'aksi: {r.action}, '
                  f'tujuan: {r.destination}, '
                  f'status: {r.status}')

if __name__ == "__main__":
    main()

Anda dapat menghapus aturan replikasi yang dikonfigurasi untuk bucket sumber untuk menonaktifkan replikasi data untuk bucket tersebut.

Contoh kode berikut menghapus aturan replikasi yang memiliki ID tertentu dari bucket yang ditentukan:

import argparse
import alibabacloud_oss_v2 as oss

# Buat parser argumen baris perintah.
parser = argparse.ArgumentParser(description="delete bucket replication sample")

# Tambahkan argumen baris perintah.
parser.add_argument('--region', help='Wilayah tempat bucket berada.', required=True)
parser.add_argument('--bucket', help='Nama bucket.', required=True)
parser.add_argument('--endpoint', help='Nama domain yang dapat digunakan layanan lain untuk mengakses OSS')
parser.add_argument('--rule_id', help='ID aturan replikasi data untuk mana Anda ingin mengonfigurasi RTC.', required=True)

def main():
    # Parsing argumen baris perintah.
    args = parser.parse_args()

    # Dapatkan kredensial akses dari variabel lingkungan.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Gunakan konfigurasi default dari SDK.
    cfg = oss.config.load_default()

    # Tentukan penyedia kredensial.
    cfg.credentials_provider = credentials_provider

    # Tentukan wilayah.
    cfg.region = args.region

    # Jika titik akhir diberikan dari baris perintah, perbarui titik akhir dalam konfigurasi dengan titik akhir yang diberikan.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Buat klien OSS.
    client = oss.Client(cfg)

    # Hapus aturan replikasi.
    result = client.delete_bucket_replication(oss.DeleteBucketReplicationRequest(
            bucket=args.bucket,
            replication_rules=oss.ReplicationRules(
                ids=[args.rule_id],
            ),
    ))

    # Tampilkan informasi respons dasar.
    print(f'kode status: {result.status_code},'
          f' ID permintaan: {result.request_id},')

if __name__ == "__main__":
    main()