Replicate data across regions by using OSS SDK for Python V2 - Object Storage Service: Central de Documentação do Alibaba Cloud

A replicação de dados copia automaticamente objetos e operações associadas, como criação, sobrescrita e exclusão, de um bucket de origem para um bucket de destino. O Object Storage Service (OSS) oferece suporte à replicação entre regiões (CRR) e à replicação na mesma região (SRR).

Observações de uso

Os códigos de exemplo neste tópico usam o ID da região cn-hangzhou, correspondente à região China (Hangzhou). Por padrão, o acesso aos recursos do bucket ocorre via endpoint público. Para acessar esses recursos a partir de outros serviços da Alibaba Cloud na mesma região do bucket, use um endpoint interno. Para mais informações sobre regiões e endpoints do OSS, consulte Regiões e endpoints.
Por padrão, uma conta Alibaba Cloud tem as permissões necessárias para replicação de dados. Para replicar dados como usuário RAM ou com credenciais de acesso temporário fornecidas pelo Security Token Service (STS), garanta que as permissões adequadas estejam configuradas.
- Para ativar a replicação de dados, é necessária a permissão oss:PutBucketReplication.
- Para ativar ou desativar o recurso de controle de tempo de replicação (RTC), é necessária a permissão oss:PutBucketRtc.
- Para consultar regras de replicação de dados, é necessária a permissão oss:GetBucketReplication.
- Para consultar as regiões de destino disponíveis para replicação de dados, é necessária a permissão oss:GetBucketReplicationLocation.
- Para consultar o progresso de uma tarefa de replicação de dados, é necessária a permissão oss:GetBucketReplicationProgress.
- Para desativar a replicação de dados, é necessária a permissão oss:DeleteBucketReplication.

Códigos de exemplo

Enable data replication

Importante

Antes de ativar a replicação de dados, certifique-se de que os buckets de origem e de destino estejam sem versionamento ou com versionamento ativado.

O código de exemplo a seguir ativa a replicação de dados e cria uma tarefa para replicar dados de um bucket de origem para um bucket de destino, na mesma região ou em uma região diferente:

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()

Query data replication rules

O código a seguir consulta as regras de replicação de dados de um 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()

Enable or disable the RTC feature

O código de exemplo a seguir ativa ou desativa o recurso RTC para uma regra 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()

Query the regions to which data can be replicated

O código de exemplo a seguir consulta as regiões de destino disponíveis para replicação de dados a partir do bucket de origem especificado:

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()

Query the progress of a data replication task

Nota

É possível consultar o progresso de tarefas de replicação de dados históricos e incrementais.

O progresso da replicação de dados históricos é expresso em porcentagem. Essa consulta está disponível apenas para buckets com replicação de dados históricos ativada.
O progresso da replicação de dados incrementais é expresso por um ponto no tempo. Os dados armazenados no bucket de origem antes desse momento já foram replicados.

O código de exemplo a seguir consulta o progresso da tarefa de replicação de dados associada a um ID específico de regra de replicação para um 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()

Disable data replication

Para desativar a replicação de dados em um bucket, exclua a regra de replicação configurada no bucket de origem.

O código de exemplo a seguir exclui a regra de replicação com um ID específico do bucket indicado:

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()