Object Storage Service:Data replication (OSS SDK for Python V1)

Quick start

The minimal end-to-end flow:

1. Set environment variables and initialize a bucket object — see Common setup.

2. (Optional) Call get_bucket_replication_location to confirm your target region is supported.

3. Call put_bucket_replication to create a replication rule.

4. Call get_bucket_replication to retrieve the assigned rule ID.

5. Call get_bucket_replication_progress to monitor replication.

6. Call delete_bucket_replication when the rule is no longer needed.

Choose a replication mode

After choosing a mode, set target_bucket_location to a region in a different Alibaba Cloud region (CRR) or the same region (SRR) when creating a replication rule.

How it works

OSS replication works as follows:

• Asynchronous replication — OSS replicates objects asynchronously after writes complete on the source bucket. There is no guaranteed replication latency; actual latency depends on object size and cross-region network distance.

• Historical data — By default, OSS replicates existing objects in addition to new writes. Set is_enable_historical_object_replication to False to replicate only objects written after the rule is created.

• Incremental data — OSS continuously replicates new writes after the rule is active.

• Delete operations — Delete operations on the source bucket are replicated to the destination. If you delete an object in the source bucket, OSS removes the corresponding object in the destination bucket.

• Encryption support — SSE-KMS-encrypted objects can be replicated with an authorized RAM role and a destination KMS key.

API overview

Prerequisites

Complete the following steps before using the SDK to manage replication rules.

Common setup

Initialize the auth and bucket objects once. All subsequent code examples in this topic reuse these objects.

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

auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
region = "cn-hangzhou"
bucket = oss2.Bucket(auth, endpoint, "examplebucket", region=region)

Replace endpoint, region, and "examplebucket" with the values for your source bucket.

Query available destination regions

Before creating a replication rule, call get_bucket_replication_location to confirm that your target region is supported.

try:
    result = bucket.get_bucket_replication_location()

    print("Available destination regions:")
    for location in result.location_list:
        print(" -", location)
except oss2.exceptions.OssError as e:
    print(f"Failed to query destination regions: {e.status} {e.request_id}")
    raise

result.location_list contains region identifiers such as oss-cn-beijing.

Related API: GetBucketReplicationLocation.

Create a replication rule

Call put_bucket_replication to create a replication rule on the source bucket. OSS assigns a rule ID at creation time; retrieve it by calling get_bucket_replication after the rule is created.

The following example creates a basic replication rule that replicates all objects to a destination bucket:

replica_config = ReplicationRule(
    target_bucket_name='destexamplebucket',
    target_bucket_location='yourTargetBucketLocation',
)

try:
    bucket.put_bucket_replication(replica_config)
    print("Replication rule created.")
except oss2.exceptions.OssError as e:
    print(f"Failed to create replication rule: {e.status} {e.request_id}")
    raise

ReplicationRule(
    target_bucket_name='string',                    # Required
    target_bucket_location='string',                # Required
    prefix_list=['prefix1', 'prefix2'],             # Optional
    action_list=[ReplicationRule.PUT],               # Optional
    is_enable_historical_object_replication=True,    # Optional (default: True)
    target_transfer_type='oss_acc',                  # Optional
    sync_role_name='string',                         # Optional
    sse_kms_encrypted_objects_status=ReplicationRule.ENABLED,  # Optional
    replica_kms_keyid='string'                       # Optional
)

# Replicate only objects with the specified prefixes.
prefix_list = ['prefix1', 'prefix2']

replica_config = ReplicationRule(
    prefix_list=prefix_list,
    # Replicate only creates and updates.
    action_list=[ReplicationRule.PUT],
    target_bucket_name='destexamplebucket',
    target_bucket_location='yourTargetBucketLocation',
    # Disable historical data replication.
    is_enable_historical_object_replication=False,
    # Use transfer acceleration.
    target_transfer_type='oss_acc',
    # Required for SSE-KMS encryption.
    sync_role_name='roleNameTest',
    sse_kms_encrypted_objects_status=ReplicationRule.ENABLED,
    replica_kms_keyid='9468da86-3509-4f8d-a61e-6eab1eac****',
)

try:
    bucket.put_bucket_replication(replica_config)
    print("Replication rule created with filtering and encryption.")
except oss2.exceptions.OssError as e:
    print(f"Failed to create replication rule: {e.status} {e.request_id}")
    raise

import time

# Upload a test object to the source bucket.
bucket.put_object('replication-test.txt', 'This is a replication test.')

# Wait for replication to complete. Replication is asynchronous and may take some time.
time.sleep(60)

# Connect to the destination bucket.
# Replace the endpoint and region with those of the destination bucket.
dest_bucket = oss2.Bucket(auth, '<destination-endpoint>', 'destexamplebucket', region='<destination-region>')

if dest_bucket.object_exists('replication-test.txt'):
    print("Replication verified: test object found in destination bucket.")
else:
    print("Object not yet replicated. Replication is asynchronous and may take some time.")

Query replication rules

Call get_bucket_replication to retrieve all replication rules configured on a bucket. Use this method to retrieve the rule ID that OSS assigns when you call put_bucket_replication.

try:
    result = bucket.get_bucket_replication()

    for rule in result.rule_list:
        print("Rule ID:", rule.rule_id)
        print("Destination bucket:", rule.target_bucket_name)
        print("Destination region:", rule.target_bucket_location)
except oss2.exceptions.OssError as e:
    print(f"Failed to query replication rules: {e.status} {e.request_id}")
    raise

Each rule in result.rule_list contains the following fields:

For the full response schema, see GetBucketReplication.

Related API: GetBucketReplication.

Query replication progress

OSS tracks replication progress in two ways:

• Historical object replication — A percentage indicating how much of the pre-existing data has been replicated. Available only when is_enable_historical_object_replication is enabled.

• Incremental replication — A point-in-time marker. All objects written to the source bucket before this timestamp have been replicated. For example, a value of "2024-01-15T12:00:00.000Z" means all objects created before that time have been replicated.

Pass the rule ID to get_bucket_replication_progress to check progress for a specific rule. Retrieve the rule ID from get_bucket_replication().

try:
    # Replace 'test_replication_1' with the rule ID returned by get_bucket_replication().
    result = bucket.get_bucket_replication_progress('test_replication_1')

    print("Rule ID:", result.progress.rule_id)
    print("Historical replication enabled:", result.progress.is_enable_historical_object_replication)
    print("Historical replication progress:", result.progress.historical_object_progress)
    print("Incremental replication progress:", result.progress.new_object_progress)
except oss2.exceptions.OssError as e:
    print(f"Failed to query replication progress: {e.status} {e.request_id}")
    raise

The result.progress object contains the following fields:

Related API: GetBucketReplicationProgress.

Delete a replication rule

Call delete_bucket_replication to delete a replication rule. Pass the rule ID to identify which rule to delete.

When you delete a replication rule:

• Objects already replicated to the destination bucket are preserved.

• Subsequent changes to the source bucket are no longer replicated.

try:
    # Replace 'test_replication_1' with the rule ID to delete.
    bucket.delete_bucket_replication('test_replication_1')
    print("Replication rule deleted successfully.")
except oss2.exceptions.OssError as e:
    print(f"Failed to delete replication rule: {e.status} {e.request_id}")
    raise

Related API: DeleteBucketReplication.

References

• PutBucketReplication

• GetBucketReplication

• GetBucketReplicationLocation

• GetBucketReplicationProgress

• DeleteBucketReplication

• Sample code: GitHub — bucket_replication.py