Carga mediante formulario con OSS SDK for Python 2.0 - Object Storage Service

La carga mediante formulario de OSS permite que las aplicaciones web carguen archivos directamente en OSS a través de formularios HTML estándar. En este tema se describe cómo usar Python SDK V2 para generar información, como firmas Post y Post Policies, y cargar archivos en OSS mediante el método HTTP POST.

Precauciones

El código de ejemplo de este tema usa el identificador de la región China (Hangzhou), cn-hangzhou. De forma predeterminada, se usa el endpoint público. Si necesita acceder a OSS desde otros productos de Alibaba Cloud en la misma región, use el endpoint interno. Para obtener más información sobre las asignaciones entre las regiones y los endpoints compatibles con OSS, consulte OSS regions and endpoints.
El tamaño de un objeto cargado mediante un formulario no puede superar los 5 GB.

Código de ejemplo

A continuación se muestra el código de ejemplo del proceso completo de una carga mediante formulario. Los pasos principales son los siguientes:

Cree una Post Policy: defina el periodo de validez y las condiciones de la solicitud de carga. Estas condiciones incluyen el nombre del bucket, la versión de la firma, la información de credenciales, la fecha de la solicitud y el rango de longitud del cuerpo de la solicitud.
Serialice y codifique la Policy: convierta la Policy en una cadena JSON y codifíquela en Base64.
Genere una clave de firma: use el algoritmo HMAC-SHA256 para generarla. La clave de firma incluye la fecha, la región, el producto y el tipo de solicitud.
Calcule la firma: firme la cadena de la Policy codificada en Base64 con la clave generada. Después, convierta el resultado en una cadena hexadecimal.
Construya el cuerpo de la solicitud: agregue al formulario la clave del objeto, la policy, la versión de la firma, la información de credenciales, la fecha de la solicitud y la firma. Luego, escriba los datos que desea cargar en el formulario.
Cree y ejecute la solicitud: genere una solicitud HTTP POST, configure el encabezado y envíela. Por último, compruebe el código de estado de la respuesta para asegurarse de que la solicitud se completó correctamente.

import argparse
import base64
import hashlib
import hmac
import json
import random
import requests
from datetime import datetime, timedelta
import alibabacloud_oss_v2 as oss

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

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

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

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

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

def main():
    # Define the content to upload.
    content = "hi oss"
    product = "oss"  # The product identifier, which is OSS.

    # Parse command-line arguments.
    args = parser.parse_args()
    region = args.region  # The region information.
    bucket_name = args.bucket  # The bucket name.
    object_name = args.key  # The object name.

    # Load credential information from environment variables for identity verification.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()
    access_key_id = credential.access_key_id  # The AccessKey ID.
    access_key_secret = credential.access_key_secret  # The AccessKey secret.

    # Obtain the current UTC time and format it.
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")

    # Set the expiration time to one hour later and create a Policy map.
    expiration = utc_time + timedelta(hours=1)
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # The expiration time of the policy.
        "conditions": [
            {"bucket": bucket_name},  # Specify the bucket.
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},  # Specify the signature version.
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},  # The credential information.
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},  # The request date.
            ["content-length-range", 1, 1024]  # The content length range limit.
        ]
    }

    # Convert the policy to a JSON string and perform Base64 encoding.
    policy = json.dumps(policy_map)
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        Build the POST request body and encode the form fields into the multipart/form-data format.
        :param field_dict: The dictionary of form fields.
        :param boundary: The separator string.
        :return: The encoded POST request body.
        """
        post_body = ''

        # Encode the form fields, except for the file content and content type.
        for k, v in field_dict.items():
            if k != 'content' and k != 'content-type':
                post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"{1}\"\r\n\r\n{2}\r\n'''.format(boundary, k, v)

        # The file content must be the last form field.
        post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'''.format(
            boundary, field_dict['content'])

        # Add the form field terminator.
        post_body += '\r\n--{0}--\r\n'.format(boundary)

        return post_body.encode('utf-8')  # Return the POST request body encoded in UTF-8.

    # Construct the signature key and use the HMAC-SHA256 algorithm to generate the signature.
    signing_key = "aliyun_v4" + access_key_secret
    h1 = hmac.new(signing_key.encode(), date.encode(), hashlib.sha256)
    h1_key = h1.digest()
    h2 = hmac.new(h1_key, region.encode(), hashlib.sha256)
    h2_key = h2.digest()
    h3 = hmac.new(h2_key, product.encode(), hashlib.sha256)
    h3_key = h3.digest()
    h4 = hmac.new(h3_key, "aliyun_v4_request".encode(), hashlib.sha256)
    h4_key = h4.digest()

    h = hmac.new(h4_key, string_to_sign.encode(), hashlib.sha256)
    signature = h.hexdigest()  # Convert the signature result to a hexadecimal string.

    # Build the dictionary of form fields required for the POST request.
    field_dict = {}
    field_dict['key'] = object_name
    field_dict['policy'] = string_to_sign
    field_dict['x-oss-signature-version'] = "OSS4-HMAC-SHA256"
    field_dict['x-oss-credential'] = f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"
    field_dict['x-oss-date'] = f"{utc_time.strftime('%Y%m%dT%H%M%SZ')}"
    field_dict['x-oss-signature'] = signature
    field_dict['content'] = content

    # Generate a random string as the form separator.
    boundary = ''.join(random.choice('0123456789') for _ in range(11))

    # Use the build_post_body function to build the POST request body.
    body = build_post_body(field_dict, boundary)

    # Construct the destination URL for the POST request.
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"

    # Set the HTTP header, specify Content-Type as multipart/form-data, and include the boundary string.
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",
    }

    # Send the POST request to OSS.
    response = requests.post(url, data=body, headers=headers)

    # Determine whether the upload is successful based on the response status code.
    if response.status_code // 100 != 2:
        print(f"Post Object Fail, status code: {response.status_code}, reason: {response.reason}")
    else:
        print(f"post object done, status code: {response.status_code}, request id: {response.headers.get('X-Oss-Request-Id')}")

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

Escenarios comunes

Carga mediante formulario con callback de carga

Si necesita notificar al servidor de aplicaciones cuando finalice una carga mediante formulario, consulte el siguiente código de ejemplo.

import argparse
import base64
import hashlib
import hmac
import json
import random
import requests
from datetime import datetime, timedelta
import alibabacloud_oss_v2 as oss

# Create a command-line argument parser to receive user-input parameters.
parser = argparse.ArgumentParser(description="post object sample")
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('--key', help='The name of the object.', required=True)
parser.add_argument('--callback_url', help='Callback server address.', required=True)

def main():
    # Define the content to upload.
    content = "hi oss"
    product = "oss"  # The product name.

    # Parse command-line arguments.
    args = parser.parse_args()
    region = args.region  # The region information.
    bucket_name = args.bucket  # The bucket name.
    object_name = args.key  # The object name (the name of the file after upload).

    # Create a credential provider using the credentials in the environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()  # Obtain the credentials.
    access_key_id = credential.access_key_id  # Obtain the AccessKey ID.
    access_key_secret = credential.access_key_secret  # Obtain the AccessKey secret.

    # Obtain the current UTC time and format it.
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")  # Format the date.
    expiration = utc_time + timedelta(hours=1)  # Set the expiration time to one hour later.

    # Build the policy document to define the upload conditions.
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # The expiration time of the policy.
        "conditions": [
            {"bucket": bucket_name},  # Specify the bucket name.
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},  # The signature version.
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},  # The credential information.
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},  # The current time.
            ["content-length-range", 1, 1024]  # The content length range limit.
        ]
    }
    # Convert the policy to a JSON string.
    policy = json.dumps(policy_map)
    # Perform Base64 encoding on the policy.
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        Build the POST request body.

        :param field_dict: The dictionary of fields.
        :param boundary: The boundary string.
        :return: The encoded request body.
        """
        post_body = ''

        # Encode the form fields.
        for k, v in field_dict.items():
            if k != 'content' and k != 'content-type':
                post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"{1}\"\r\n\r\n{2}\r\n'''.format(boundary, k, v)

        # The content of the uploaded file must be the last form field.
        post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'''.format(
            boundary, field_dict['content'])

        # Add the form field terminator.
        post_body += '\r\n--{0}--\r\n'.format(boundary)

        return post_body.encode('utf-8')

    # Build the signature key.
    signing_key = "aliyun_v4" + access_key_secret
    h1 = hmac.new(signing_key.encode(), date.encode(), hashlib.sha256)
    h1_key = h1.digest()
    h2 = hmac.new(h1_key, region.encode(), hashlib.sha256)
    h2_key = h2.digest()
    h3 = hmac.new(h2_key, product.encode(), hashlib.sha256)
    h3_key = h3.digest()
    h4 = hmac.new(h3_key, "aliyun_v4_request".encode(), hashlib.sha256)
    h4_key = h4.digest()

    h = hmac.new(h4_key, string_to_sign.encode(), hashlib.sha256)
    signature = h.hexdigest()  # Calculate the HMAC-SHA256 signature.

    # Build the dictionary of form fields.
    field_dict = {}
    field_dict['key'] = object_name  # The object name.
    field_dict['policy'] = string_to_sign  # The policy.
    field_dict['x-oss-signature-version'] = "OSS4-HMAC-SHA256"  # The signature version.
    field_dict['x-oss-credential'] = f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"  # The credential information.
    field_dict['x-oss-date'] = f"{utc_time.strftime('%Y%m%dT%H%M%SZ')}"  # The current time.
    field_dict['x-oss-signature'] = signature  # The signature value.
    field_dict['content'] = content  # The file content.

    def encode_callback(callback_params):
        """
        Perform Base64 encoding on the callback parameters.

        :param callback_params: The dictionary of callback parameters.
        :return: The Base64-encoded string.
        """
        cb_str = json.dumps(callback_params).strip()
        return base64.b64encode(cb_str.encode()).decode()

    # Set the upload callback parameters.
    callback_params = {}
    callback_params['callbackUrl'] = args.callback_url  # The callback server address.
    callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'  # The callback request body.
    callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'  # The callback request body type.
    encoded_callback = encode_callback(callback_params)  # Encode the callback parameters.

    # Add callback-related fields to the form data.
    field_dict['callback'] = encoded_callback
    field_dict['x:my_var1'] = 'value1'
    field_dict['x:my_var2'] = 'value2'

    # Generate a random boundary string.
    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    # Send the POST request.
    body = build_post_body(field_dict, boundary)

    # Build the URL for the OSS service.
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",  # Set the request header.
    }

    # Send the POST request.
    response = requests.post(url, data=body, headers=headers)

    # Process the response.
    if response.status_code // 100 != 2:
        print(f"Post Object Fail, status code: {response.status_code}, reason: {response.reason}")
    else:
        print(f"post object done, status code: {response.status_code}, request id: {response.headers.get('X-Oss-Request-Id')}")

    # Print the response content.
    print(f"response: {response.text}")

if __name__ == "__main__":
    main()

Referencias

Para consultar un ejemplo completo de carga mediante formulario, revise el archivo post_object.py.