Object Storage Service:Upload callbacks (Python SDK V1)

Usage notes

• The examples in this topic use the public endpoint for the China (Hangzhou) region. To access OSS from other Alibaba Cloud services in the same region, use an internal endpoint. For more information, see Regions and endpoints.

• The examples create an OSSClient instance using an OSS endpoint. To create an OSSClient instance with a custom domain name or Security Token Service (STS), see Initialization.

How it works

All three upload types follow the same callback flow:

1. Build the callback parameters — specify the callback URL, request body, and any custom variables.

2. JSON-serialize and Base64-encode the parameters, then attach them to the upload request using x-oss-callback and x-oss-callback-var.

3. After the upload completes, OSS sends a POST request to your callback URL.

Simple upload callback

The following example uploads an object and triggers a callback using put_object.

# -*- coding: utf-8 -*-
import json
import base64
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# Load credentials from environment variables OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Set the endpoint for the region where your bucket is located.
# Example: https://oss-cn-hangzhou.aliyuncs.com for China (Hangzhou).
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Set the region ID. Required for V4 signatures.
region = "cn-hangzhou"

bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)


def encode_callback(params):
    """JSON-serialize params and Base64-encode the result.
    OSS requires callback parameters in this format.
    """
    cb_str = json.dumps(params).strip()
    return oss2.compat.to_string(base64.b64encode(oss2.compat.to_bytes(cb_str)))


# Build the callback parameters.
callback_params = {
    # The URL of your callback server.
    'callbackUrl': 'http://oss-demo.aliyuncs.com:23450',
    # (Optional) Override the Host header in the callback request.
    # 'callbackHost': 'yourCallbackHost',
    # The callback request body. Use ${variable} placeholders for upload metadata.
    'callbackBody': 'bucket=${bucket}&object=${object}&size=${size}'
                    '&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}',
    'callbackBodyType': 'application/x-www-form-urlencoded',
}
encoded_callback = encode_callback(callback_params)

# Custom variables. Keys must start with x:.
callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
encoded_callback_var = encode_callback(callback_var_params)

# Attach the callback parameters to the upload request.
params = {
    'x-oss-callback': encoded_callback,
    'x-oss-callback-var': encoded_callback_var,
}

# Upload the object with callback enabled.
result = bucket.put_object('examplefiles/exampleobject.txt', 'a' * 1024 * 1024, params)

Form upload callback

Form upload uses the PostObject API and does not require the OSS SDK. Pass the callback parameters as a Base64-encoded form field named callback.

# -*- coding: utf-8 -*-
import os
import time
import datetime
import json
import base64
import hmac
import hashlib
import crcmod
import requests


# Load credentials and configuration from environment variables.
access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<Your AccessKey ID>')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<Your AccessKey secret>')
bucket_name = os.getenv('OSS_TEST_BUCKET', '<Your bucket name>')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '<Your endpoint>')
call_back_url = "http://oss-demo.aliyuncs.com:23450"
region = "<Your region>"  # Example: cn-hangzhou
product = "oss"

# Verify that all required parameters are set.
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<' not in param, 'Set the parameter: ' + param


def convert_base64(input):
    return base64.b64encode(input.encode(encoding='utf-8')).decode('utf-8')


def calculate_crc64(data):
    """Calculate the CRC-64 hash of the data."""
    _POLY = 0x142F0E1EBA9EA3693
    _XOROUT = 0XFFFFFFFFFFFFFFFF

    crc64 = crcmod.Crc(_POLY, initCrc=0, xorOut=_XOROUT)
    crc64.update(data.encode())

    return crc64.crcValue


def build_gmt_expired_time(expire_time):
    """Generate the request expiration time in ISO 8601 format.

    :param int expire_time: The timeout period in seconds.
    :return str: The expiration time.
    """
    now = int(time.time())
    expire_syncpoint = now + expire_time
    expire_gmt = datetime.datetime.fromtimestamp(expire_syncpoint).isoformat()
    expire_gmt += 'Z'
    return expire_gmt


def build_encode_policy(expired_time, condition_list):
    """Build and Base64-encode the upload policy.

    :param int expired_time: The expiration time in seconds.
    :param list condition_list: The list of policy conditions.
    """
    policy_dict = {
        'expiration': build_gmt_expired_time(expired_time),
        'conditions': condition_list,
    }
    policy = json.dumps(policy_dict).strip()
    return base64.b64encode(policy.encode())


def build_signature(access_key_secret, date):
    """Derive the V4 signing key and compute the request signature.

    The key derivation 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)
    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)
    return h.hexdigest()


def build_callback(cb_url, cb_body, cb_body_type=None, cb_host=None):
    """Build and Base64-encode the callback string.

    :param str cb_url: The URL of your callback server.
    :param str cb_body: The callback request body.
    :param str cb_body_type: The Content-Type of the callback request.
        Defaults to application/x-www-form-urlencoded.
    :param str cb_host: The value of the Host header in the callback request.
    :return str: The Base64-encoded callback string.
    """
    callback_dict = {'callbackUrl': cb_url, 'callbackBody': cb_body}
    callback_dict['callbackBodyType'] = cb_body_type or 'application/x-www-form-urlencoded'
    if cb_host is not None:
        callback_dict['callbackHost'] = cb_host

    callback_param = json.dumps(callback_dict).strip()
    return base64.b64encode(callback_param.encode()).decode()


def build_post_url(endpoint, bucket_name):
    """Build the PostObject URL: https://{bucket}.{endpoint}."""
    if endpoint.startswith('http://'):
        return endpoint.replace('http://', f'http://{bucket_name}.')
    elif endpoint.startswith('https://'):
        return endpoint.replace('https://', f'https://{bucket_name}.')
    return f'http://{bucket_name}.{endpoint}'


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

    The file content field must be last.
    """
    post_body = ''
    for k, v in field_dict.items():
        if k not in ('content', 'content-type'):
            post_body += f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'

    post_body += (
        f'--{boundary}\r\n'
        f'Content-Disposition: form-data; name="file"; filename="{field_dict["key"]}"\r\n'
        f'Content-Type: {field_dict["content-type"]}\r\n\r\n'
        f'{field_dict["content"]}'
    )
    post_body += f'\r\n--{boundary}--\r\n'
    return post_body.encode('utf-8')


def build_post_headers(body_len, boundary, headers=None):
    """Build the request headers for the PostObject request."""
    headers = headers if headers else {}
    headers['Content-Length'] = str(body_len)
    headers['Content-Type'] = f'multipart/form-data; boundary={boundary}'
    return headers


# Build the form fields. Field names are case-sensitive.
utc_time = datetime.datetime.utcnow()
date = utc_time.strftime("%Y%m%d")
expiration = '2120-01-01T12:00:00.000Z'

policy_map = {
    "expiration": expiration,
    "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()

field_dict = {
    'key': '0303/post.txt',                      # Object path in OSS
    'OSSAccessKeyId': access_key_id,
    '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': build_signature(access_key_secret, date),
    # (Optional) STS token. Required when using temporary credentials.
    # 'x-oss-security-token': '',
    'Content-Disposition': 'attachment;filename=download.txt',
    'x-oss-meta-uuid': 'uuid-xxx',               # User-defined metadata
    # Callback parameters. Remove these fields if you do not need a callback.
    'callback': build_callback(
        call_back_url,
        'bucket=${bucket}&object=${object}&size=${size}'
        '&mimeType=${mimeType}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}',
        'application/x-www-form-urlencoded',
    ),
    # Custom callback variables.
    'x:my_var1': 'value1',
    'x:my_var2': 'value2',
    # File content. To upload an actual file, read its content here.
    # with open("<file_path>", "rb") as f:
    #     field_dict['content'] = f.read()
    'content': 'a' * 64,
    'content-type': 'text/plain',
}

boundary = '9431149156168'

body = build_post_body(field_dict, boundary)
headers = build_post_headers(len(body), boundary)

resp = requests.post(build_post_url(endpoint, bucket_name), data=body, headers=headers)

print(resp.status_code)
assert resp.status_code == 200
assert resp.headers['x-oss-hash-crc64ecma'] == str(calculate_crc64(field_dict['content']))

What's next

• For the complete sample code, see the GitHub example.

• For the upload callback API reference, see Callback.

• For more information about PostObject, see PostObject.