Upload callback (Python SDK V1)

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

# Obtain access credentials from environment variables. Before you run this example, make sure you have configured the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Set endpoint to the endpoint of the region where your bucket is located. For example, if your bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Set region to the region that corresponds to your endpoint, for example, cn-hangzhou. Note: This parameter is required for Signature V4.
region = "cn-hangzhou"

# Set yourBucketName to the name of your bucket.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# Define a function to Base64-encode the callback parameters.
def encode_callback(callback_params):
    cb_str = json.dumps(callback_params).strip()
    return oss2.compat.to_string(base64.b64encode(oss2.compat.to_bytes(cb_str)))

# Set the upload callback parameters.
callback_params = {}
# Set the URL of your callback server. For example, http://oss-demo.aliyuncs.com:23450.
callback_params['callbackUrl'] = 'http://oss-demo.aliyuncs.com:23450'
# (Optional) Set the Host value in the callback request header. This should match the Host configured on your server.
#callback_params['callbackHost'] = 'yourCallbackHost'
# Specify the body of the callback request. Use placeholders to dynamically pass object information.
callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Specify the Content-Type of the callback request.
callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'
encoded_callback = encode_callback(callback_params)
# Set custom parameters for the callback request. Each parameter is a key-value pair, and the key 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)

# Prepare the parameters for the upload with a callback.
params = {'x-oss-callback': encoded_callback, 'x-oss-callback-var': encoded_callback_var}
# Specify the full path and content of the object. Do not include the bucket name in the full path.
result = bucket.put_object('examplefiles/exampleobject.txt', 'a'*1024*1024, params)

# -*- coding: utf-8 -*-

import json
from oss2.credentials import EnvironmentVariableCredentialsProvider
import oss2

key = 'exampleobject.txt'
content = "Anything you're good at contributes to happiness."

# Obtain access credentials from environment variables. Before you run this example, make sure you have configured the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Set endpoint to the endpoint of the region where your bucket is located. For example, if your bucket is in the China (Hangzhou) region, set the endpoint to https://oss-cn-hangzhou.aliyuncs.com.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Set region to the region that corresponds to your endpoint, for example, cn-hangzhou. Note: This parameter is required for Signature V4.
region = "cn-hangzhou"

# Set the URL of your callback server. For example, http://oss-demo.aliyuncs.com:23450.
callback_url = 'http://oss-demo.aliyuncs.com:23450'

# Set yourBucketName to the name of your bucket.
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)


# Prepare the callback parameters.
callback_dict = {}
callback_dict['callbackUrl'] = callback_url

# (Optional) Set the Host value in the callback request header. This should match the Host configured on your server.
# callback_dict['callbackHost'] = 'oss-cn-hangzhou.aliyuncs.com'

# Specify the body of the callback request. Use placeholders to dynamically pass object information.
callback_dict['callbackBody'] = 'bucket=${bucket}&object=${object}&size=${size}&mimeType=${mimeType}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'
# Specify the Content-Type of the callback request.
callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded'

callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
callback_var_param_json = json.dumps(callback_var_params).strip()
encoded_callback_var = oss2.utils.b64encode_as_string(callback_var_param_json)

# Callback parameters must be JSON formatted and Base64 encoded.
callback_param = json.dumps(callback_dict).strip()
base64_callback_body = oss2.utils.b64encode_as_string(callback_param)
# Place the encoded callback parameters in the request header.
headers = {'x-oss-callback': base64_callback_body, 'x-oss-callback-var': encoded_callback_var}


"""
Multipart upload callback
"""

# Perform a multipart upload with a callback.
# Initialize the multipart upload task.
parts = []
upload_id = bucket.init_multipart_upload(key).upload_id
# Upload a part.
result = bucket.upload_part(key, upload_id, 1, content)
parts.append(oss2.models.PartInfo(1, result.etag, size = len(content), part_crc = result.crc))
# Complete the upload and trigger the callback.
result = bucket.complete_multipart_upload(key, upload_id, parts, headers)

# A status code of 200 means both the upload and callback were successful. A status code of 203 means the upload was successful, but the callback failed.
if result.status == 200:
    print("Upload successful, and callback successful (HTTP 200)")
elif result.status == 203:
    print("Upload successful, but callback failed (HTTP 203)")
else:
    print(f"Upload failed with status code: {result.status}")


# Verify that the file was uploaded successfully.
result = bucket.head_object(key)
assert 'x-oss-hash-crc64ecma' in result.headers

For more information about PostObject, see PostObject and Troubleshoot PostObject errors.

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


# The following code shows how to use PostObject. This implementation does not depend on the OSS SDK for Python.

# First, initialize parameters such as your AccessKey ID, AccessKey Secret, and endpoint.
# Obtain these from environment variables or replace the placeholders with your actual values.
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>')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '<your endpoint>')
# This example uses http://oss-demo.aliyuncs.com:23450.
call_back_url = "http://oss-demo.aliyuncs.com:23450"
# For example, cn-hangzhou.
region = "<your region>"


# Make sure the preceding parameters are correctly configured.
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<' not in param, 'Set the following parameter: ' + param

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

def calculate_crc64(data):
    """Calculate the CRC64 hash of the data.
    :param data: The data.
    :return: The CRC64 value 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 a request expiration time in GMT format.
    :param int expire_time: The timeout period in seconds.
    :return str: The expiration time in GMT format.
    """
    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):
    """Generate a policy.
    :param int expired_time: The expiration time in seconds.
    :param list condition_list: The list of conditions.
    """
    policy_dict = {}
    policy_dict['expiration'] = build_gmt_expired_time(expired_time)
    policy_dict['conditions'] = condition_list

    policy = json.dumps(policy_dict).strip()
    policy_encode = base64.b64encode(policy.encode())

    return policy_encode

def build_signature(access_key_secret, date):
    """Generate a signature.
    :param str access_key_secret: The AccessKey Secret.
    :return str: The request 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()

    return signature

def build_callback(cb_url, cb_body, cb_body_type=None, cb_host=None):
    """Generate a callback string.
    :param str cb_url: The URL of the callback server. OSS sends a callback request to this URL after the file is uploaded.
    :param str cb_body: The body of the callback request.
    :param str cb_body_type: The Content-Type of the callback request. Default value: application/x-www-form-urlencoded.
    :param str cb_host: The value of the Host header in the callback request.
    :return str: The encoded callback string.
    """
    callback_dict = {}

    callback_dict['callbackUrl'] = cb_url

    callback_dict['callbackBody'] = cb_body
    if cb_body_type is None:
        callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded'
    else:
        callback_dict['callbackBodyType'] = cb_body_type

    if cb_host is not None:
        callback_dict['callbackHost'] = cb_host

    callback_param = json.dumps(callback_dict).strip()
    base64_callback = base64.b64encode(callback_param.encode());

    return base64_callback.decode()

def build_post_url(endpoint, bucket_name):
    """Generate a URL for the POST request.
    :param str endpoint: The endpoint.
    :param str bucket_name: The bucket name.
    :return str: The URL for the POST request.
    """
    if endpoint.startswith('http://'):
        return endpoint.replace('http://', 'http://{0}.'.format(bucket_name))
    elif endpoint.startswith('https://'):
        return endpoint.replace('https://', 'https://{0}.'.format(bucket_name))
    else:
        return 'http://{0}.{1}'.format(bucket_name, endpoint)

def build_post_body(field_dict, boundary):
    """Generate a body for the POST request.
    :param dict field_dict: The form fields for the POST request.
    :param str boundary: The boundary string for the form fields.
    :return str: The body for the POST request.
    """
    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 file to upload, which must be the last form field.
    post_body += '''--{0}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{1}\"\r\nContent-Type: {2}\r\n\r\n{3}'''.format(
        boundary, field_dict['key'], field_dict['content-type'], field_dict['content'])

    # Add the closing boundary.
    post_body += '\r\n--{0}--\r\n'.format(boundary)

    return post_body.encode('utf-8')

def build_post_headers(body_len, boundary, headers=None):
    """Generate headers for the POST request.
    :param str body_len: The length of the POST request body.
    :param str boundary: The boundary string of the form fields.
    :param dict headers: The request headers.
    """
    headers = headers if headers else {}
    headers['Content-Length'] = str(body_len)
    headers['Content-Type'] = 'multipart/form-data; boundary={0}'.format(boundary)

    return headers

def encode_callback(callback_params):
    cb_str = json.dumps(callback_params).strip()
    return base64.b64encode(cb_str.encode()).decode()

# The form fields for the POST request are case-sensitive.
field_dict = {}
# The name of the object.
field_dict['key'] = '0303/post.txt'
# The AccessKey ID.
field_dict['OSSAccessKeyId'] = access_key_id

product = "oss"


utc_time = datetime.datetime.utcnow()
# Set the policy expiration time.
expiration = '2120-01-01T12:00:00.000Z'
date = utc_time.strftime("%Y%m%d")
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)
print(policy)
string_to_sign = base64.b64encode(policy.encode()).decode()

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')}"
# The request signature.
field_dict['x-oss-signature'] = build_signature(access_key_secret, date)



# The STS token. Required only when using temporary credentials. Otherwise, leave this empty or remove it.
# field_dict['x-oss-security-token'] = ''
# Content-Disposition
field_dict['Content-Disposition'] = 'attachment;filename=download.txt'
# Custom user metadata.
field_dict['x-oss-meta-uuid'] = 'uuid-xxx'
# The callback. Do not include this field if you do not require a callback.
field_dict['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 variables for the callback. Do not include this field if you do not require a callback.
field_dict['x:my_var1'] = 'value1'
field_dict['x:my_var2'] = 'value2'

# To upload a file:
# with open("", r) as f:
#     content = f.read()
# field_dict['content'] = content

# The content to upload.
field_dict['content'] = 'a'*64
# The content type of the uploaded file.
field_dict['content-type'] = 'text/plain'

# The boundary string for the form fields, which is typically a random string.
boundary = '9431149156168'

# Send the POST request.
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)

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