Carga multiparte (SDK de Python V2) - Object Storage Service - Centro de documentación de Alibaba Cloud

La función de carga multiparte de Object Storage Service (OSS) le permite dividir un objeto grande en varias partes. Una vez que se cargan estas partes, puede llamar a la operación CompleteMultipartUpload para combinarlas en un objeto completo.

Notas

El código de ejemplo de este tema usa de forma predeterminada la región China (Hangzhou) (cn-hangzhou) y su endpoint público. Si desea acceder a OSS desde otros productos de Alibaba Cloud en la misma región, use un endpoint interno. Para obtener más información sobre las regiones y los endpoints compatibles con OSS, consulte Regions and endpoints.
Para ejecutar una carga multiparte, debe contar con el permiso oss:PutObject. Para obtener más información, consulte Grant a RAM user custom permissions.

Proceso de carga multiparte

Una carga multiparte consta de los tres pasos a continuación:

Inicialice un evento de carga multiparte.

Llame al método Client.InitiateMultipartUpload para obtener de OSS un ID de carga único globalmente.
Cargue las partes.

Llame al método Client.UploadPart para cargar los datos de cada parte.
Nota
- Para un mismo ID de carga, el número de parte identifica la parte y su posición relativa dentro del objeto completo. Si carga una parte nueva con un número de parte ya existente, se sobrescribe la parte existente en OSS.
- OSS incluye el hash MD5 de los datos de la parte recibida en el encabezado ETag de la respuesta.
- OSS calcula el hash MD5 de los datos cargados y lo compara con el hash MD5 que calculó el kit de desarrollo de software (SDK). Si ambos hashes MD5 son distintos, se devuelve el código de error InvalidDigest.
Complete la carga multiparte.

Después de cargar todas las partes, llame al método Client.CompleteMultipartUpload para fusionarlas en un objeto completo.

Código de ejemplo

A continuación se muestra un código de ejemplo que divide un archivo local grande en varias partes, las carga de manera simultánea en un bucket y luego las fusiona en un objeto completo.

import os
import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser for the multipart upload sample.
parser = argparse.ArgumentParser(description="multipart upload sample")

# Add the required --region command-line argument, which specifies the region where the bucket is located.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the required --bucket command-line argument, which specifies the name of the bucket.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the optional --endpoint command-line argument, which specifies the domain name that other services can use to access OSS.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the required --key command-line argument, which specifies the name of the object.
parser.add_argument('--key', help='The name of the object.', required=True)

# Add the required --file_path command-line argument, which specifies the path of the file to upload.
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

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

    # Load credentials from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configurations of the SDK and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set the region in the configuration.
    cfg.region = args.region

    # If an endpoint is provided, set the endpoint in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client based on the configuration.
    client = oss.Client(cfg)

    # Initiate a multipart upload request to obtain the upload ID for subsequent part uploads.
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # Define the size of each part as 5 MB.
    part_size = 5 * 1024 * 1024

    # Obtain the total size of the file to upload.
    data_size = os.path.getsize(args.file_path)

    # Initialize the part number, starting from 1.
    part_number = 1

    # Store the result of each part upload.
    upload_parts = []

    # Open the file in binary read mode.
    with open(args.file_path, 'rb') as f:
        # Traverse the file and upload it in parts based on part_size.
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # Handle the case where the last part may be smaller than part_size.
                n = data_size - start

            # Create a SectionReader to read a specific portion of the file.
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # Upload the part.
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader
            ))

            # Print the result of each part upload.
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # Save the part upload result to the list.
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # Increment the part number.
            part_number += 1

    # Sort the uploaded parts by part number.
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # Send a request to complete the multipart upload and merge all parts into a complete object.
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        )
    ))

    # The following code provides another method to list and merge all part data into a complete object on the server.
    # This method is suitable when you are not sure whether all parts are successfully uploaded.
    # Merge fragmented data into a complete Object through the server-side List method
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # Print the result of the completed multipart upload.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # The script entry point. The main function is called when the file is run directly.

Escenarios comunes

Ejecutar una carga multiparte y configurar callbacks de carga

Si necesita notificar a un servidor de aplicaciones cuando finalice una carga multiparte, consulte el siguiente código de ejemplo.

import os
import argparse
import base64
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser for the multipart upload sample.
parser = argparse.ArgumentParser(description="multipart upload sample")

# Add the required --region command-line argument, which specifies the region where the bucket is located.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the required --bucket command-line argument, which specifies the name of the bucket.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the optional --endpoint command-line argument, which specifies the domain name that other services can use to access OSS.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the required --key command-line argument, which specifies the name of the object.
parser.add_argument('--key', help='The name of the object.', required=True)

# Add the required --file_path command-line argument, which specifies the path of the file to upload.
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

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

    # Load credentials from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configurations of the SDK and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set the region in the configuration.
    cfg.region = args.region

    # If an endpoint is provided, set the endpoint in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client based on the configuration.
    client = oss.Client(cfg)

    # Initiate a multipart upload request to obtain the upload ID for subsequent part uploads.
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # Define the size of each part as 1 MB.
    part_size = 1 * 1024 * 1024

    # Obtain the total size of the file to upload.
    data_size = os.path.getsize(args.file_path)

    # Initialize the part number, starting from 1.
    part_number = 1

    # Store the result of each part upload.
    upload_parts = []

    # Open the file in binary read mode.
    with open(args.file_path, 'rb') as f:
        # Traverse the file and upload it in parts based on part_size.
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # Handle the case where the last part may be smaller than part_size.
                n = data_size - start

            # Create a SectionReader to read a specific portion of the file.
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # Upload the part.
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader
            ))

            # Print the result of each part upload.
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # Save the part upload result to the list.
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # Increment the part number.
            part_number += 1

    # Sort the uploaded parts by part number.
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # Define the webhook address.
    call_back_url = "http://www.example.com/callback"
    # Construct the callback parameter: specify the webhook address and the request body, and encode them in Base64.
    callback=base64.b64encode(str('{\"callbackUrl\":\"' + call_back_url + '\",\"callbackBody\":\"bucket=${bucket}&object=${object}&my_var_1=${x:var1}&my_var_2=${x:var2}\"}').encode()).decode()
    # Construct the custom variables (callback-var) and encode them in Base64.
    callback_var=base64.b64encode('{\"x:var1\":\"value1\",\"x:var2\":\"value2\"}'.encode()).decode()

    # Send a request to complete the multipart upload and merge all parts into a complete object.
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        ),
        callback=callback,
        callback_var=callback_var
    ))

    # The following code provides another method to list and merge all part data into a complete object on the server.
    # This method is suitable when you are not sure whether all parts are successfully uploaded.
    # Merge fragmented data into a complete Object through the server-side List method
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # Print the result of the completed multipart upload.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # The script entry point. The main function is called when the file is run directly.

Mostrar una barra de progreso para una carga multiparte

import os
import argparse
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser for the multipart upload sample.
parser = argparse.ArgumentParser(description="multipart upload sample")

# Add the required --region command-line argument, which specifies the region where the bucket is located.
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# Add the required --bucket command-line argument, which specifies the name of the bucket.
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# Add the optional --endpoint command-line argument, which specifies the domain name that other services can use to access OSS.
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# Add the required --key command-line argument, which specifies the name of the object.
parser.add_argument('--key', help='The name of the object.', required=True)

# Add the required --file_path command-line argument, which specifies the path of the file to upload.
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

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

    # Load credentials from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Use the default configurations of the SDK and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Set the region in the configuration.
    cfg.region = args.region

    # If an endpoint is provided, set the endpoint in the configuration.
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # Create an OSS client based on the configuration.
    client = oss.Client(cfg)

    # Define a dictionary variable named progress_state to save the upload progress. The initial value is 0.
    progress_state = {'saved': 0}
    def _progress_fn(n, written, total):
        # Use a dictionary to store the accumulated written bytes to avoid using global variables.
        progress_state['saved'] += n

        # Calculate the current upload percentage. Divide the number of written bytes by the total number of bytes and round down the result.
        rate = int(100 * (float(written) / float(total)))

        # Print the current upload progress. \r returns the cursor to the beginning of the line to enable real-time refresh in the command line.
        # end='' prevents a new line, which allows the next print to overwrite the current line.
        print(f'\rUpload progress: {rate}% ', end='')

    # Initiate a multipart upload request to obtain the upload ID for subsequent part uploads.
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # Define the size of each part as 5 MB.
    part_size = 5 * 1024 * 1024

    # Obtain the total size of the file to upload.
    data_size = os.path.getsize(args.file_path)

    # Initialize the part number, starting from 1.
    part_number = 1

    # Store the result of each part upload.
    upload_parts = []

    # Open the file in binary read mode.
    with open(args.file_path, 'rb') as f:
        # Traverse the file and upload it in parts based on part_size.
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # Handle the case where the last part may be smaller than part_size.
                n = data_size - start

            # Create a SectionReader to read a specific portion of the file.
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # Upload the part.
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader,
                progress_fn=_progress_fn
            ))

            # Print the result of each part upload.
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # Save the part upload result to the list.
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # Increment the part number.
            part_number += 1

    # Sort the uploaded parts by part number.
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # Send a request to complete the multipart upload and merge all parts into a complete object.
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        )
    ))

    # The following code provides another method to list and merge all part data into a complete object on the server.
    # This method is suitable when you are not sure whether all parts are successfully uploaded.
    # Merge fragmented data into a complete Object through the server-side List method
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # Print the result of the completed multipart upload.
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # The script entry point. The main function is called when the file is run directly.

Referencias

Para consultar el código de ejemplo completo de la carga multiparte, consulte complete_multipart_upload.py.