Como usar o OSS SDK for Python 2.0 para upload de formulário - Object Storage Service: Central de Documentação do Alibaba Cloud

O upload de formulário do OSS permite que aplicações web enviem arquivos diretamente para o OSS por meio de formulários HTML padrão. Este tópico descreve como usar o Python SDK V2 para gerar informações, como assinaturas Post e Post Policies, e enviar arquivos ao OSS mediante chamada ao método HTTP POST.

Precauções

Os códigos de exemplo deste tópico usam o ID da região China (Hangzhou), cn-hangzhou, como referência. O endpoint público é o padrão. Para acessar o OSS a partir de outros produtos da Alibaba Cloud na mesma região, use o endpoint interno. Para mais informações sobre os mapeamentos entre regiões e endpoints compatíveis com o OSS, consulte Regiões e endpoints do OSS.
O tamanho de um objeto enviado via upload de formulário não pode exceder 5 GB.

Código de exemplo

O código a seguir demonstra o processo completo de upload de formulário. As principais etapas são:

Crie uma Post Policy: defina o período de validade e as condições da solicitação de upload. Essas condições incluem o nome do bucket, a versão da assinatura, as informações de credencial, a data da solicitação e o intervalo de tamanho do corpo da requisição.
Serialize e codifique a Policy: serialize a Policy em uma string JSON e codifique-a em Base64.
Gere uma chave de assinatura: utilize o algoritmo HMAC-SHA256 para gerar a chave de assinatura. A chave deve incluir a data, a região, o produto e o tipo de solicitação.
Calcule a assinatura: use a chave gerada para assinar a string da Policy codificada em Base64 e converta o resultado em uma string hexadecimal.
Monte o corpo da requisição: adicione a chave do objeto, a policy, a versão da assinatura, as informações de credencial, a data da solicitação e a assinatura ao formulário. Em seguida, insira os dados a serem enviados.
Crie e execute a requisição: crie uma requisição HTTP POST, defina o cabeçalho e envie a requisição. Por fim, verifique o código de status da resposta para confirmar o sucesso da operação.

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.

Cenários comuns

Upload de formulário com callback de upload

Para notificar o servidor da aplicação após a conclusão do upload de formulário, consulte o código de exemplo a seguir.

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

Referências

Para obter um exemplo completo de upload de formulário, consulte post_object.py.