Object Storage Service:Form upload (Python SDK V2)

Usage notes

• The sample code uses cn-hangzhou as the region. The public endpoint is used by default. To access OSS from another Alibaba Cloud service in the same region, use the internal endpoint. For region-to-endpoint mappings, see OSS regions and endpoints.

• The maximum object size for form upload is 5 GB.

How it works

A form upload involves two phases that typically run on separate machines:

Phase 1 — Server side: generate the Post Policy and signature

1. Create a Post Policy — a JSON document that defines the upload conditions: bucket name, signature version, credential information, request date, and content length range.

2. Serialize the policy to a JSON string, then Base64-encode it.

3. Derive a signing key using HMAC-SHA256 chained over: aliyun_v4 + AccessKey secret → date → region → product → aliyun_v4_request.

4. Sign the Base64-encoded policy string with the derived key, and convert the result to a hexadecimal string.

Phase 2 — Client side: submit the form

1. Build a multipart/form-data request body containing the object key, encoded policy, signature version, credential information, request date, and signature. The file content must be the last field.

2. Send an HTTP POST request to http://{bucket}.oss-{region}.aliyuncs.com. A 2xx response indicates success.

Note: If you are building a browser-based upload form, map the generated fields to <input type="hidden"> elements. For example:

<form action="http://<bucket>.oss-<region>.aliyuncs.com" method="post" enctype="multipart/form-data">
  <input type="hidden" name="key" value="<object-key>" />
  <input type="hidden" name="policy" value="<base64-encoded-policy>" />
  <input type="hidden" name="x-oss-signature-version" value="OSS4-HMAC-SHA256" />
  <input type="hidden" name="x-oss-credential" value="<credential>" />
  <input type="hidden" name="x-oss-date" value="<utc-datetime>" />
  <input type="hidden" name="x-oss-signature" value="<signature>" />
  <input type="file" name="file" />
  <input type="submit" value="Upload to OSS" />
</form>

Sample code

The following example shows the complete form upload flow — from generating the Post Policy and signature on the server side to submitting the POST request.

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

# Parse command-line arguments.
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)


def build_post_body(field_dict, boundary):
    """
    Build the multipart/form-data POST request body.

    :param field_dict: Form fields as a dictionary.
    :param boundary: Boundary string separating each field.
    :return: The encoded request body as bytes.
    """
    post_body = ''

    # Encode all fields except 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 field.
    post_body += '--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'.format(
        boundary, field_dict['content'])
    post_body += '\r\n--{0}--\r\n'.format(boundary)

    return post_body.encode('utf-8')


def main():
    content = "hi oss"
    product = "oss"

    args = parser.parse_args()
    region = args.region
    bucket_name = args.bucket
    object_name = args.key

    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()
    access_key_id = credential.access_key_id
    access_key_secret = credential.access_key_secret

    # Get the current UTC time.
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")

    # --- Phase 1: Generate the Post Policy and signature (server side) ---

    # Step 1: Create the Post Policy.
    # Set expiration to one hour from now and define upload conditions.
    expiration = utc_time + timedelta(hours=1)
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
        "conditions": [
            {"bucket": bucket_name},
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},
            ["content-length-range", 1, 1024]  # Allow objects between 1 and 1024 bytes.
        ]
    }

    # Step 2: Serialize the policy to JSON and Base64-encode it.
    policy = json.dumps(policy_map)
    string_to_sign = base64.b64encode(policy.encode()).decode()

    # Step 3: Derive the signing key using HMAC-SHA256.
    # Chain: aliyun_v4+secret -> date -> region -> product -> aliyun_v4_request
    signing_key = "aliyun_v4" + access_key_secret
    h1 = hmac.new(signing_key.encode(), date.encode(), hashlib.sha256)
    h2 = hmac.new(h1.digest(), region.encode(), hashlib.sha256)
    h3 = hmac.new(h2.digest(), product.encode(), hashlib.sha256)
    h4 = hmac.new(h3.digest(), "aliyun_v4_request".encode(), hashlib.sha256)

    # Step 4: Sign the Base64-encoded policy and convert to a hexadecimal string.
    h = hmac.new(h4.digest(), string_to_sign.encode(), hashlib.sha256)
    signature = h.hexdigest()

    # --- Phase 2: Submit the form (client side) ---

    # Step 5: Build the form fields.
    field_dict = {
        'key': object_name,
        'policy': string_to_sign,
        'x-oss-signature-version': "OSS4-HMAC-SHA256",
        'x-oss-credential': f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request",
        'x-oss-date': utc_time.strftime('%Y%m%dT%H%M%SZ'),
        'x-oss-signature': signature,
        'content': content,
    }

    # Generate an 11-digit random boundary string.
    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    body = build_post_body(field_dict, boundary)

    # Step 6: Send the POST request and check the response.
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {"Content-Type": f"multipart/form-data; boundary={boundary}"}
    response = requests.post(url, data=body, headers=headers)

    if response.status_code // 100 != 2:
        print(f"Post Object failed. Status code: {response.status_code}, reason: {response.reason}")
    else:
        print(f"Post Object succeeded. Status code: {response.status_code}, request ID: {response.headers.get('X-Oss-Request-Id')}")


if __name__ == "__main__":
    main()

Replace the following placeholders when running the sample:

Common scenarios

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

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 build_post_body(field_dict, boundary):
    """
    Build the multipart/form-data POST request body.

    :param field_dict: Form fields as a dictionary.
    :param boundary: Boundary string separating each field.
    :return: The encoded request body as bytes.
    """
    post_body = ''
    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)
    post_body += '--{0}\r\nContent-Disposition: form-data; name=\"file\"\r\n\r\n{1}'.format(
        boundary, field_dict['content'])
    post_body += '\r\n--{0}--\r\n'.format(boundary)
    return post_body.encode('utf-8')


def encode_callback(callback_params):
    """
    Base64-encode the callback parameters for inclusion in the form.

    :param callback_params: Callback configuration as a dictionary.
    :return: Base64-encoded string.
    """
    cb_str = json.dumps(callback_params).strip()
    return base64.b64encode(cb_str.encode()).decode()


def main():
    content = "hi oss"
    product = "oss"

    args = parser.parse_args()
    region = args.region
    bucket_name = args.bucket
    object_name = args.key

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    credential = credentials_provider.get_credentials()
    access_key_id = credential.access_key_id
    access_key_secret = credential.access_key_secret

    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")
    expiration = utc_time + timedelta(hours=1)

    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),
        "conditions": [
            {"bucket": bucket_name},
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},
            {"x-oss-credential": f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request"},
            {"x-oss-date": utc_time.strftime("%Y%m%dT%H%M%SZ")},
            ["content-length-range", 1, 1024]
        ]
    }
    policy = json.dumps(policy_map)
    string_to_sign = base64.b64encode(policy.encode()).decode()

    signing_key = "aliyun_v4" + access_key_secret
    h1 = hmac.new(signing_key.encode(), date.encode(), hashlib.sha256)
    h2 = hmac.new(h1.digest(), region.encode(), hashlib.sha256)
    h3 = hmac.new(h2.digest(), product.encode(), hashlib.sha256)
    h4 = hmac.new(h3.digest(), "aliyun_v4_request".encode(), hashlib.sha256)
    h = hmac.new(h4.digest(), string_to_sign.encode(), hashlib.sha256)
    signature = h.hexdigest()

    field_dict = {
        'key': object_name,
        'policy': string_to_sign,
        'x-oss-signature-version': "OSS4-HMAC-SHA256",
        'x-oss-credential': f"{access_key_id}/{date}/{region}/{product}/aliyun_v4_request",
        'x-oss-date': utc_time.strftime('%Y%m%dT%H%M%SZ'),
        'x-oss-signature': signature,
        'content': content,
    }

    # Configure the upload callback.
    # OSS sends a POST request to callbackUrl with the specified body after upload.
    callback_params = {
        'callbackUrl': args.callback_url,
        'callbackBody': 'bucket=${bucket}&object=${object}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}',
        'callbackBodyType': 'application/x-www-form-urlencoded',
    }
    field_dict['callback'] = encode_callback(callback_params)

    # Add custom variables to pass through the callback body.
    field_dict['x:my_var1'] = 'value1'
    field_dict['x:my_var2'] = 'value2'

    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    body = build_post_body(field_dict, boundary)

    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {"Content-Type": f"multipart/form-data; boundary={boundary}"}
    response = requests.post(url, data=body, headers=headers)

    if response.status_code // 100 != 2:
        print(f"Post Object failed. Status code: {response.status_code}, reason: {response.reason}")
    else:
        print(f"Post Object succeeded. Status code: {response.status_code}, request ID: {response.headers.get('X-Oss-Request-Id')}")

    print(f"Response: {response.text}")


if __name__ == "__main__":
    main()

References

• For a complete runnable example, see post_object.py.