OSS フォームアップロードを使用すると、Web アプリケーションは標準の HTML フォームを使用してファイルを OSS に直接アップロードできます。 このトピックでは、Python SDK V2 を使用して Post 署名や Post ポリシーなどの情報を生成し、HTTP POST メソッドを呼び出すことによってファイルを OSS にアップロードする方法について説明します。
注意事項
このトピックのサンプルコードでは、中国 (杭州) リージョン ID
cn-hangzhouを例として使用しています。 デフォルトではパブリックエンドポイントが使用されます。 同じリージョン内の他の Alibaba Cloud プロダクトから OSS にアクセスする場合は、内部エンドポイントを使用します。 OSS でサポートされているリージョンとエンドポイントのマッピングの詳細については、「OSS のリージョンとエンドポイント」をご参照ください。フォームアップロードを使用してアップロードされるオブジェクトのサイズは 5 GB を超えることはできません。
サンプルコード
次のサンプルコードは、フォームアップロードの完全なプロセスを示しています。主な手順は次のとおりです。
Post ポリシー を作成する: アップロードリクエストの有効期間と条件を定義します。 条件には、バケット名、署名バージョン、認証情報、リクエスト日時、リクエスト本文の長さの範囲が含まれます。
ポリシーをシリアル化してエンコードする: ポリシーを JSON 文字列にシリアル化し、Base64 を使用してエンコードします。
署名キーを生成する: HMAC-SHA256 アルゴリズムを使用して署名キーを生成します。 署名キーには、日付、リージョン、プロダクト、リクエストタイプが含まれます。
署名を計算する: 生成されたキーを使用して、Base64 エンコードされたポリシー文字列に署名し、署名結果を 16 進数文字列に変換します。
リクエスト本文を作成する: オブジェクトキー、ポリシー、署名バージョン、認証情報、リクエスト日時、署名をフォームに追加します。 次に、アップロードするデータをフォームに書き込みます。
リクエストを作成して実行する: HTTP POST リクエストを作成し、リクエストヘッダーを設定してリクエストを送信します。 次に、レスポンスステータスコードを確認して、リクエストが成功したことを確認します。
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
# POST オブジェクトアップロードサンプルのコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="post object sample")
# バケットが配置されているリージョンを指定するコマンドライン引数 --region を追加します。このパラメーターは必須です。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
# バケットの名前を指定するコマンドライン引数 --bucket を追加します。このパラメーターは必須です。
parser.add_argument('--bucket', help='バケットの名前。', required=True)
# 他のサービスが OSS にアクセスするために使用できるドメイン名を指定するコマンドライン引数 --endpoint を追加します。このパラメーターはオプションです。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
# オブジェクトの名前を指定するコマンドライン引数 --key を追加します。このパラメーターは必須です。
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
def main():
# アップロードするコンテンツを定義します。
content = "hi oss"
product = "oss" # プロダクト識別子。OSS です。
# コマンドライン引数を解析します。
args = parser.parse_args()
region = args.region # リージョン情報。
bucket_name = args.bucket # バケット名。
object_name = args.key # オブジェクト名。
# ID 検証のために環境変数から認証情報をロードします。
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
credential = credentials_provider.get_credentials()
access_key_id = credential.access_key_id # AccessKey ID。
access_key_secret = credential.access_key_secret # AccessKey シークレット。
# 現在の UTC 時刻を取得してフォーマットします。
utc_time = datetime.utcnow()
date = utc_time.strftime("%Y%m%d")
# 有効期限を 1 時間後に設定し、ポリシーマップを作成します。
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] # コンテンツの長さの範囲制限。
]
}
# ポリシーを JSON 文字列に変換し、Base64 エンコーディングを実行します。
policy = json.dumps(policy_map)
string_to_sign = base64.b64encode(policy.encode()).decode()
def build_post_body(field_dict, boundary):
"""
POST リクエスト本文を作成し、フォームフィールドを multipart/form-data 形式にエンコードします。
:param field_dict: フォームフィールドの辞書。
:param boundary: 区切り文字列。
:return: エンコードされた POST リクエスト本文。
"""
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') # UTF-8 でエンコードされた POST リクエスト本文を返します。
# 署名キーを構築し、HMAC-SHA256 アルゴリズムを使用して署名を生成します。
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() # 署名結果を 16 進数文字列に変換します。
# POST リクエストに必要なフォームフィールドの辞書を作成します。
field_dict = {}
field_dict['key'] = object_name
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')}"
field_dict['x-oss-signature'] = signature
field_dict['content'] = content
# ランダムな文字列をフォームセパレータとして生成します。
boundary = ''.join(random.choice('0123456789') for _ in range(11))
# build_post_body 関数を使用して POST リクエスト本文を作成します。
body = build_post_body(field_dict, boundary)
# POST リクエストの宛先 URL を構築します。
url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
# HTTP ヘッダーを設定し、Content-Type を multipart/form-data として指定し、境界文字列を含めます。
headers = {
"Content-Type": f"multipart/form-data; boundary={boundary}",
}
# POST リクエストを OSS に送信します。
response = requests.post(url, data=body, headers=headers)
# レスポンスステータスコードに基づいてアップロードが成功したかどうかを判断します。
if response.status_code // 100 != 2:
print(f"Post Object Fail, status code: {response.status_code}, reason: {response.reason}")
else:
print(f"post object done, status code: {response.status_code}, request id: {response.headers.get('X-Oss-Request-Id')}")
if __name__ == "__main__":
main() # スクリプトのエントリポイント。ファイルが直接実行されると、main 関数が呼び出されます。一般的なシナリオ
関連情報
フォームアップロードの完全な例については、「post_object.py」を参照してください。