Répliquer des données entre régions à l'aide du SDK OSS pour Python V2 - Object Storage Service

La réplication de données copie automatiquement les objets et les opérations associées (création, écrasement et suppression) d'un bucket source vers un bucket de destination. Object Storage Service (OSS) prend en charge la réplication inter-régions (CRR) et la réplication intra-région (SRR).

Remarques

L'exemple de code de cette rubrique utilise l'ID de région cn-hangzhou de la région Chine (Hangzhou). Par défaut, le point de terminaison public permet d'accéder aux ressources d'un bucket. Pour accéder aux ressources du bucket depuis d'autres services Alibaba Cloud situés dans la même région, utilisez un point de terminaison interne. Pour plus d'informations sur les régions et les points de terminaison OSS, consultez la rubrique Régions et points de terminaison.
Par défaut, un compte Alibaba Cloud dispose des autorisations nécessaires à la réplication de données. Pour répliquer des données en tant qu'utilisateur RAM ou via des identifiants d'accès temporaires fournis par Security Token Service (STS), vous devez disposer des autorisations requises.
- Pour activer la réplication de données, l'autorisation oss:PutBucketReplication est requise.
- Pour activer ou désactiver la fonctionnalité de contrôle du temps de réplication (RTC), l'autorisation oss:PutBucketRtc est requise.
- Pour interroger les règles de réplication de données, l'autorisation oss:GetBucketReplication est requise.
- Pour interroger les régions de destination disponibles pour la réplication de données, l'autorisation oss:GetBucketReplicationLocation est requise.
- Pour interroger la progression d'une tâche de réplication de données, l'autorisation oss:GetBucketReplicationProgress est requise.
- Pour désactiver la réplication de données, l'autorisation oss:DeleteBucketReplication est requise.

Exemples de code

Activer la réplication de données

Important

Avant d'activer la réplication de données, vérifiez que le bucket source et le bucket de destination sont tous deux dans un état sans versioning ou avec versioning activé.

Le code suivant active la réplication de données et crée une tâche qui réplique les données d'un bucket source vers un bucket de destination situé dans la même région ou dans une région différente :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="put bucket replication sample")
# Specify the command-line arguments.
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')
parser.add_argument('--sync_role', help='The role that you want to authorize OSS to use to replicate data', required=True)
parser.add_argument('--target_bucket', help='The destination bucket to which data is replicated', required=True)
parser.add_argument('--target_location', help='The region in which the destination bucket is located', required=True)

def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

    # Obtain access credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()
    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider
    # Specify the region.
    cfg.region = args.region
    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Configure and execute the PutBucketReplication request.
    result = client.put_bucket_replication(oss.PutBucketReplicationRequest(
            bucket=args.bucket,  # The name of the source bucket.
            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',  # Disable RTC.
                    ),
                    destination=oss.ReplicationDestination(
                        bucket=args.target_bucket,  # The name of the destination bucket.
                        location=args.target_location,  # The region of the destination bucket.
                        transfer_type=oss.TransferType.INTERNAL,  # The transfer type.
                    ),
                    historical_object_replication=oss.HistoricalObjectReplicationType.DISABLED,  # Forbid replication of historical data.
                    sync_role=args.sync_role,  # The role for data replication.
                    status='Disabled',  # Disable the rule.
                    prefix_set=oss.ReplicationPrefixSet(
                        prefixs=['aaa/', 'bbb/'],  # The prefixes in the names of objects to be replicated.
                    ),
                    action='ALL',  # All operations.
                )],
            ),
    ))

    # Display the HTTP status code and request ID.
    print(f'status code: {result.status_code}, request id: {result.request_id}')

if __name__ == "__main__":
    main()

Interroger les règles de réplication de données

Le code suivant interroge les règles de réplication de données d'un bucket :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="get bucket replication sample")
# Add the command-line arguments.
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():
    # Parse the command-line arguments.
    args = parser.parse_args()

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

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()
    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider
    # Specify the region.
    cfg.region = args.region
    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Query the replication rules for the bucket.
    result = client.get_bucket_replication(oss.GetBucketReplicationRequest(
            bucket=args.bucket,
    ))

    # Display the basic response information.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' replication configuration: {result.replication_configuration}')

    # If there are data replication rules, display rule details.
    if result.replication_configuration.rules:
        for r in result.replication_configuration.rules:
            print(f'result: source selection criteria: {r.source_selection_criteria}, '
                  f'rtc: {r.rtc}, destination: {r.destination}, '
                  f'historical object replication: {r.historical_object_replication}, '
                  f'sync role: {r.sync_role}, status: {r.status}, '
                  f'encryption configuration: {r.encryption_configuration}, '
                  f'id: {r.id}, prefix set: {r.prefix_set}, action: {r.action}')

if __name__ == "__main__":
    main()

Activer ou désactiver la fonctionnalité RTC

Le code suivant active ou désactive la fonctionnalité RTC pour une règle CRR :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="put bucket rtc sample")

# Add the command-line arguments.
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')
parser.add_argument('--status', help='Specifies whether to enable RTC. Valid values: disabled, enabled', default='disabled')
parser.add_argument('--rule_id', help='The ID of the data replication rule for which you want to configure RTC.', required=True)

def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

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

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()
    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider
    # Specify the region.
    cfg.region = args.region
    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Create an RTS state configuration and apply it to the specified bucket.
    result = client.put_bucket_rtc(oss.PutBucketRtcRequest(
            bucket=args.bucket,
            rtc_configuration=oss.RtcConfiguration(
                rtc=oss.ReplicationTimeControl(
                    status=args.status,
                ),
                id=args.rule_id,
            ),
    ))

    # Display the basic response information.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id}')

if __name__ == "__main__":
    main()

Interroger les régions vers lesquelles les données peuvent être répliquées

Le code suivant interroge les régions vers lesquelles les données peuvent être répliquées depuis le bucket source spécifié :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="get bucket replication location sample")
# Add the command-line arguments.
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():
    # Parse the command-line arguments.
    args = parser.parse_args()

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

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()
    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider
    # Specify the region.
    cfg.region = args.region
    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Query the regions to which data in the specified bucket can be replicated.
    result = client.get_bucket_replication_location(oss.GetBucketReplicationLocationRequest(
            bucket=args.bucket,
    ))

    # Display the basic response information.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' replication location: {result.replication_location},'
          f' location transfer type constraint: {result.replication_location.location_transfer_type_constraint},'
          # f' location: {result.replication_location.location_transfer_type_constraint.location_transfer_types[0].location},'
          # f' transfer types: {result.replication_location.location_transfer_type_constraint.location_transfer_types[0].transfer_types},'
          # f' location: {result.replication_location.location_transfer_type_constraint.location_transfer_types[1].location},'
          # f' transfer types: {result.replication_location.location_transfer_type_constraint.location_transfer_types[1].transfer_types},'
          f' locationrtc constraint: {result.replication_location.locationrtc_constraint},'
    )

    # If restrictions on destination regions for data replication exist, display region details and transfer types.
    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'result: location: {r.location}, transfer types: {r.transfer_types}')

if __name__ == "__main__":
    main()

Interroger la progression d'une tâche de réplication de données

Remarque

Il est possible d'interroger la progression des tâches de réplication de données historiques et incrémentielles.

La progression d'une tâche de réplication de données historiques est exprimée en pourcentage. Cette information n'est disponible que pour les buckets dont la réplication de données historiques est activée.
La progression d'une tâche de réplication de données incrémentielles est exprimée par un horodatage. Les données stockées dans le bucket source avant cet instant sont répliquées.

Le code suivant interroge la progression de la tâche de réplication de données associée à un ID de règle de réplication spécifique pour un bucket :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="get bucket replication progress sample")
# Add the command-line arguments.
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')
parser.add_argument('--rule_id', help='The ID of the data replication rule for which you want to configure RTC.', required=True)

def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

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

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()
    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider
    # Specify the region.
    cfg.region = args.region
    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Query the data replication progress.
    result = client.get_bucket_replication_progress(oss.GetBucketReplicationProgressRequest(
            bucket=args.bucket,
            rule_id=args.rule_id,
    ))

    # Display the basic response information.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' replication progress: {result.replication_progress}')

    # If the data replication rule exists, display rule details.
    if result.replication_progress.rules:
        for r in result.replication_progress.rules:
            print(f'result: historical object replication: {r.historical_object_replication}, '
                  f'progress: {r.progress}, '
                  f'id: {r.id}, '
                  f'prefix set: {r.prefix_set}, '
                  f'action: {r.action}, '
                  f'destination: {r.destination}, '
                  f'status: {r.status}')

if __name__ == "__main__":
    main()

Désactiver la réplication de données

Supprimez la règle de réplication configurée pour un bucket source afin de désactiver la réplication de données pour ce bucket.

Le code suivant supprime la règle de réplication dotée d'un ID spécifique du bucket indiqué :

import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser.
parser = argparse.ArgumentParser(description="delete bucket replication sample")

# Add the command-line arguments.
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')
parser.add_argument('--rule_id', help='The ID of the data replication rule for which you want to configure RTC.', required=True)

def main():
    # Parse the command-line arguments.
    args = parser.parse_args()

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

    # Use the default configuration of the SDK.
    cfg = oss.config.load_default()

    # Specify the credential provider.
    cfg.credentials_provider = credentials_provider

    # Specify the region.
    cfg.region = args.region

    # If an endpoint is provided from the command line, update the endpoint in the configuration with the provided endpoint.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client.
    client = oss.Client(cfg)

    # Delete the replication rule.
    result = client.delete_bucket_replication(oss.DeleteBucketReplicationRequest(
            bucket=args.bucket,
            replication_rules=oss.ReplicationRules(
                ids=[args.rule_id],
            ),
    ))

    # Display the basic response information.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},')

if __name__ == "__main__":
    main()