Utilisation du SDK OSS pour Python 2.0 pour le téléchargement via formulaire - Object Storage Service

Le téléchargement via formulaire OSS permet aux applications web de transférer des fichiers directement vers OSS en utilisant des formulaires HTML standard. Cette rubrique explique comment utiliser la version 2 du SDK Python pour générer les informations nécessaires, telles que les signatures Post et les politiques Post, et pour télécharger des fichiers vers OSS en appelant la méthode HTTP POST.

Précautions

L'exemple de code de cette rubrique utilise l'ID de région Chine (Hangzhou), cn-hangzhou, à titre d'exemple. L'endpoint public est utilisé par défaut. Si vous souhaitez accéder à OSS depuis d'autres produits Alibaba Cloud dans la même région, utilisez l'endpoint interne. Pour plus d'informations sur les correspondances entre les régions et les endpoints pris en charge par OSS, consultez Régions et endpoints OSS.
La taille d'un objet téléchargé via un formulaire ne peut pas dépasser 5 Go.

Exemple de code

L'exemple de code suivant illustre le processus complet d'un téléchargement via formulaire. Les étapes principales sont les suivantes :

Créez une Post Policy : définissez la période de validité et les conditions de la demande de téléchargement. Les conditions incluent le nom du bucket, la version de la signature, les informations d'identification, la date de la demande et la plage de longueur du corps de la demande.
Sérialisez et encodez la politique : sérialisez la politique dans une chaîne JSON, puis encodez-la en Base64.
Générez une clé de signature : utilisez l'algorithme HMAC-SHA256 pour générer une clé de signature. La clé de signature comprend la date, la région, le produit et le type de demande.
Calculez la signature : utilisez la clé générée pour signer la chaîne de politique encodée en Base64, puis convertissez le résultat de la signature en une chaîne hexadécimale.
Construisez le corps de la demande : ajoutez la clé de l'objet, la politique, la version de la signature, les informations d'identification, la date de la demande et la signature au formulaire. Ensuite, écrivez les données à télécharger dans le formulaire.
Créez et exécutez la demande : créez une demande HTTP POST, définissez l'en-tête de la demande et envoyez-la. Vérifiez ensuite le code d'état de la réponse pour vous assurer que la demande a abouti.

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.

Scénarios courants

Téléchargement via formulaire avec un callback de téléchargement

Si vous souhaitez notifier le serveur d'application une fois le téléchargement via formulaire terminé, reportez-vous à l'exemple de code suivant.

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

Références

Pour un exemple complet de téléchargement via formulaire, consultez post_object.py.