Object Storage Service:Upload callbacks (Python SDK V1)

注意事項

• このトピックの例では、中国 (杭州) リージョンのパブリックエンドポイントを使用します。同じリージョン内の他の Alibaba Cloud サービスから OSS にアクセスする場合は、内部エンドポイントを使用します。詳細については、「リージョンとエンドポイント」をご参照ください。

• この例では、OSS エンドポイントを使用して OSSClient インスタンスを作成します。カスタムドメイン名またはセキュリティトークンサービス (STS) を使用して OSSClient インスタンスを作成するには、初期化をご参照ください。

仕組み

3 種類のアップロードはすべて、同じコールバックフローに従います。

1. コールバックパラメーターの構築 — コールバック URL、リクエストボディ、およびカスタム変数を指定します。

2. パラメーターを JSON シリアル化し、Base64 エンコードしてから、x-oss-callback および x-oss-callback-var を使用してアップロードリクエストにアタッチします。

3. アップロードが完了すると、OSS はご利用のコールバック URL に POST リクエストを送信します。

シンプルアップロードコールバック

次の例では、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.
# 環境変数 OSS_ACCESS_KEY_ID および 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).
# 例: 中国 (杭州) の場合は https://oss-cn-hangzhou.aliyuncs.com。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Set the region ID. Required for V4 signatures.
# リージョン ID を設定します。V4 署名に必須です。
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.
    """
    # パラメーターを JSON シリアル化し、結果を Base64 エンコードします。
    # OSS はこの形式のコールバックパラメーターを必要とします。
    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.
    # ご利用のコールバックサーバーの URL。
    'callbackUrl': 'http://oss-demo.aliyuncs.com:23450',
    # (Optional) Override the Host header in the callback request.
    # (オプション) コールバックリクエストの Host ヘッダーを上書きします。
    # 'callbackHost': 'yourCallbackHost',
    # The callback request body. Use ${variable} placeholders for upload metadata.
    # コールバックリクエストボディ。アップロードメタデータには ${variable} プレースホルダーを使用します。
    '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:.
# カスタム変数。キーは 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)

フォームアップロードコールバック

フォームアップロードでは、PostObject API を使用し、OSS SDK は必要ありません。コールバックパラメーターを、callback という名前の Base64 エンコード済みフォームフィールドとして渡します。

# -*- 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."""
    # データの CRC-64 ハッシュを計算します。
    _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.
    """
    # ISO 8601 形式でリクエストの有効期限を生成します。
    # :param int expire_time: タイムアウト期間 (秒単位)。
    # :return str: 有効期限。
    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.
    """
    # アップロードポリシーを構築し、Base64 エンコードします。
    # :param int expired_time: 有効期限 (秒単位)。
    # :param list condition_list: ポリシー条件のリスト。
    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
    """
    # V4 署名キーを派生させ、リクエスト署名を計算します。
    # キー派生チェーン: 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.
    """
    # コールバック文字列を構築し、Base64 エンコードします。
    # :param str cb_url: ご利用のコールバックサーバーの URL。
    # :param str cb_body: コールバックリクエストボディ。
    # :param str cb_body_type: コールバックリクエストの Content-Type。
    #     デフォルトは application/x-www-form-urlencoded です。
    # :param str cb_host: コールバックリクエストの Host ヘッダーの値。
    # :return str: Base64 エンコードされたコールバック文字列。
    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}."""
    # 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.
    """
    # multipart/form-data ボディを構築します。
    # ファイルコンテンツフィールドは最後に配置する必要があります。
    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."""
    # PostObject リクエストのリクエストヘッダーを構築します。
    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
    # 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.
    # (オプション) STS トークン。一時的な認証情報を使用する場合に必須です。
    # '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']))

次のステップ

• 完全なサンプルコードについては、「GitHub の例」をご参照ください。

• アップロードコールバック API リファレンスについては、「コールバック」をご参照ください。

• PostObject の詳細については、「PostObject」をご参照ください。