OSS表单上传允许网页应用通过标准HTML表单直接将文件上传至OSS。本文介绍如何使用Python SDK V2生成Post签名和Post Policy等信息,并调用HTTP Post方法上传文件到OSS。
注意事项
本文示例代码以华东1(杭州)的地域ID
cn-hangzhou
为例,默认使用外网Endpoint,如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系,请参见OSS地域和访问域名。通过表单上传的方式上传的Object大小不能超过5 GB。
示例代码
以下代码示例实现了表单上传的完整过程,主要步骤如下:
创建Post Policy:定义上传请求的有效时间和条件,包括存储桶名称、签名版本、凭证信息、请求日期和请求体长度范围。
序列化并编码Policy:将Policy序列化为JSON字符串,并进行Base64编码。
生成签名密钥:使用HMAC-SHA256算法生成签名密钥,包括日期、区域、产品和请求类型。
计算签名:使用生成的密钥对Base64编码后的Policy字符串进行签名,并将签名结果转换为十六进制字符串。
构建请求体:添加对象键、策略、签名版本、凭证信息、请求日期和签名到表单中,并将要上传的数据写入表单。
创建并执行请求:创建一个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='The region in which the bucket is located.', required=True)
# 添加命令行参数 --bucket,表示存储空间的名称,必需参数
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
# 添加命令行参数 --endpoint,表示其他服务可用来访问OSS的域名,非必需参数
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
# 添加命令行参数 --key,表示对象的名称,必需参数
parser.add_argument('--key', help='The name of the object.', 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 # 对象名称
# 从环境变量中加载凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
credential = credentials_provider.get_credentials()
access_key_id = credential.access_key_id # 访问密钥ID
access_key_secret = credential.access_key_secret # 访问密钥秘密
# 获取当前UTC时间并格式化
utc_time = datetime.utcnow()
date = utc_time.strftime("%Y%m%d")
# 设置过期时间为1小时后,并创建策略(Policy)映射
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() # 签名结果转换为十六进制字符串
# 构建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。
该文章对您有帮助吗?