Object Storage Service:Upload por formulário usando OSS SDK for Python 2.0

Precauções

• O código de exemplo neste tópico usa o ID de região China (Hangzhou), cn-hangzhou, como referência. Por padrão, o endpoint público é utilizado. Para acessar o OSS a partir de outros produtos 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 OSS regions and endpoints.

• O tamanho de um objeto enviado via upload por formulário não pode exceder 5 GB.

Código de exemplo

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

1. Crie uma Post Policy: defina o período de validade e as condições da requisição de upload. As condições incluem o nome do bucket, a versão da assinatura, as informações de credencial, a data da requisição e o intervalo de tamanho do corpo da requisição.

2. Serialize e codifique a Policy: serialize a Policy em uma string JSON e codifique-a em Base64.

3. Gere uma chave de assinatura: use o algoritmo HMAC-SHA256 para gerar a chave de assinatura, que inclui data, região, product e tipo de requisição.

4. Calcule a assinatura: use a chave gerada para assinar a string da Policy codificada em Base64 e converta o resultado em uma string hexadecimal.

5. Monte o corpo da requisição: adicione ao formulário a chave do objeto, a policy, a versão da assinatura, as informações de credencial, a data da requisição e a assinatura. Em seguida, grave os dados a serem enviados no formulário.

6. Crie e execute a requisição: crie uma requisição HTTP POST, defina o cabeçalho da requisição e envie-a. Verifique o código de status da resposta para confirmar que a requisição foi bem-sucedida.

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

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 um exemplo completo de upload por formulário, consulte post_object.py.