Object Storage Service:Callback unggahan (Python SDK V1)

Catatan penggunaan

• Contoh dalam topik ini menggunakan titik akhir publik untuk wilayah China (Hangzhou). Untuk mengakses OSS dari layanan Alibaba Cloud lainnya di wilayah yang sama, gunakan titik akhir internal. Untuk informasi selengkapnya, lihat Regions and endpoints.

• Contoh membuat instans OSSClient menggunakan titik akhir OSS. Untuk membuat instans OSSClient dengan nama domain kustom atau Security Token Service (STS), lihat Initialization.

Cara kerja

Ketiga jenis unggahan mengikuti alur callback yang sama:

1. Buat parameter callback—tentukan URL callback, badan permintaan, dan variabel kustom apa pun.

2. Serialisasi ke JSON dan encode dengan Base64, lalu sambungkan ke permintaan unggahan menggunakan x-oss-callback dan x-oss-callback-var.

3. Setelah unggahan selesai, OSS mengirimkan permintaan POST ke URL callback Anda.

Callback unggahan simple

Contoh berikut mengunggah objek dan memicu callback menggunakan put_object.

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

# Muat kredensial dari variabel lingkungan OSS_ACCESS_KEY_ID dan OSS_ACCESS_KEY_SECRET.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Tetapkan titik akhir untuk wilayah tempat bucket Anda berada.
# Contoh: https://oss-cn-hangzhou.aliyuncs.com untuk China (Hangzhou).
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# Tetapkan ID wilayah. Diperlukan untuk signature V4.
region = "cn-hangzhou"

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


def encode_callback(params):
    """Serialisasi params ke JSON dan encode hasilnya dengan Base64.
    OSS mengharuskan parameter callback dalam format ini.
    """
    cb_str = json.dumps(params).strip()
    return oss2.compat.to_string(base64.b64encode(oss2.compat.to_bytes(cb_str)))


# Bangun parameter callback.
callback_params = {
    # URL server callback Anda.
    'callbackUrl': 'http://oss-demo.aliyuncs.com:23450',
    # (Opsional) Timpa header Host dalam permintaan callback.
    # 'callbackHost': 'yourCallbackHost',
    # Badan permintaan callback. Gunakan placeholder ${variable} untuk metadata unggahan.
    '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)

# Variabel kustom. Kunci harus diawali dengan x:.
callback_var_params = {'x:my_var1': 'my_val1', 'x:my_var2': 'my_val2'}
encoded_callback_var = encode_callback(callback_var_params)

# Sambungkan parameter callback ke permintaan unggahan.
params = {
    'x-oss-callback': encoded_callback,
    'x-oss-callback-var': encoded_callback_var,
}

# Unggah objek dengan callback diaktifkan.
result = bucket.put_object('examplefiles/exampleobject.txt', 'a' * 1024 * 1024, params)

Callback unggahan formulir

Unggahan formulir menggunakan API PostObject dan tidak memerlukan OSS SDK. Teruskan parameter callback sebagai field formulir yang diencode Base64 dengan nama callback.

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


# Muat kredensial dan konfigurasi dari variabel lingkungan.
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>"  # Contoh: cn-hangzhou
product = "oss"

# Verifikasi bahwa semua parameter yang diperlukan telah ditetapkan.
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<' not in param, 'Setel parameter: ' + param


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


def calculate_crc64(data):
    """Hitung hash CRC-64 dari 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):
    """Hasilkan waktu kedaluwarsa permintaan dalam format ISO 8601.

    :param int expire_time: Periode timeout dalam detik.
    :return str: Waktu kedaluwarsa.
    """
    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):
    """Bangun dan encode kebijakan unggahan dengan Base64.

    :param int expired_time: Waktu kedaluwarsa dalam detik.
    :param list condition_list: Daftar kondisi kebijakan.
    """
    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):
    """Turunkan kunci penandatanganan V4 dan hitung signature permintaan.

    Rantai turunan kunci: 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):
    """Bangun dan encode string callback dengan Base64.

    :param str cb_url: URL server callback Anda.
    :param str cb_body: Badan permintaan callback.
    :param str cb_body_type: Content-Type permintaan callback.
        Default-nya adalah application/x-www-form-urlencoded.
    :param str cb_host: Nilai header Host dalam permintaan callback.
    :return str: String callback yang diencode 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):
    """Bangun URL PostObject: 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):
    """Bangun badan multipart/form-data.

    Field konten file harus berada di akhir.
    """
    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):
    """Bangun header permintaan untuk permintaan PostObject."""
    headers = headers if headers else {}
    headers['Content-Length'] = str(body_len)
    headers['Content-Type'] = f'multipart/form-data; boundary={boundary}'
    return headers


# Bangun field formulir. Nama field bersifat 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',                      # Path objek di 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),
    # (Opsional) token STS. Diperlukan saat menggunakan kredensial temporary.
    # 'x-oss-security-token': '',
    'Content-Disposition': 'attachment;filename=download.txt',
    'x-oss-meta-uuid': 'uuid-xxx',               # Metadata yang ditentukan pengguna
    # Parameter callback. Hapus field ini jika Anda tidak memerlukan 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',
    ),
    # Variabel callback kustom.
    'x:my_var1': 'value1',
    'x:my_var2': 'value2',
    # Konten file. Untuk mengunggah file aktual, baca kontennya di sini.
    # 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']))

Langkah berikutnya

• Untuk kode contoh lengkap, lihat contoh GitHub.

• Untuk referensi API callback unggahan, lihat Callback.

• Untuk informasi selengkapnya tentang PostObject, lihat PostObject.