表单上传(Python SDK V2)

OSS表单上传允许网页应用通过标准HTML表单直接将文件上传至OSS。本文介绍如何使用Python SDK V2生成Post签名和Post Policy等信息,并调用HTTP Post方法上传文件到OSS。

注意事项

  • 本文示例代码以华东1(杭州)的地域IDcn-hangzhou为例,默认使用外网Endpoint,如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的RegionEndpoint的对应关系,请参见OSS地域和访问域名

  • 通过表单上传的方式上传的Object大小不能超过5 GB。

示例代码

以下代码示例实现了表单上传的完整过程,主要步骤如下:

  1. 创建Post Policy:定义上传请求的有效时间和条件,包括存储桶名称、签名版本、凭证信息、请求日期和请求体长度范围。

  2. 序列化并编码Policy:将Policy序列化为JSON字符串,并进行Base64编码。

  3. 生成签名密钥:使用HMAC-SHA256算法生成签名密钥,包括日期、区域、产品和请求类型。

  4. 计算签名:使用生成的密钥对Base64编码后的Policy字符串进行签名,并将签名结果转换为十六进制字符串。

  5. 构建请求体:添加对象键、策略、签名版本、凭证信息、请求日期和签名到表单中,并将要上传的数据写入表单。

  6. 创建并执行请求:创建一个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函数

常见使用场景

表单上传并设置上传回调

如果您希望在表单上传后通知应用服务器,可参考以下代码示例。

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

# 创建命令行参数解析器,用于接收用户输入的参数
parser = argparse.ArgumentParser(description="post object sample")
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--key', help='The name of the object.', required=True)
parser.add_argument('--callback_url', help='Callback server address.', required=True)

def main():
    # 定义要上传的内容
    content = "hi oss"
    product = "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  # 获取Access Key ID
    access_key_secret = credential.access_key_secret  # 获取Access Key Secret

    # 获取当前UTC时间并格式化
    utc_time = datetime.utcnow()
    date = utc_time.strftime("%Y%m%d")  # 格式化日期
    expiration = utc_time + timedelta(hours=1)  # 设置过期时间为1小时后

    # 构建策略文档,定义上传条件
    policy_map = {
        "expiration": expiration.strftime("%Y-%m-%dT%H:%M:%S.000Z"),  # 策略过期时间
        "conditions": [
            {"bucket": bucket_name},  # 指定Bucket名称
            {"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字符串
    policy = json.dumps(policy_map)
    # 对策略进行Base64编码
    string_to_sign = base64.b64encode(policy.encode()).decode()

    def build_post_body(field_dict, boundary):
        """
        构建POST请求体。

        :param field_dict: 字段字典
        :param boundary: 边界字符串
        :return: 编码后的请求体
        """
        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')

    # 构建签名密钥
    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()  # 计算HMAC-SHA256签名

    # 构建表单字段字典
    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  # 文件内容

    def encode_callback(callback_params):
        """
        对回调参数进行Base64编码。

        :param callback_params: 回调参数字典
        :return: Base64编码后的字符串
        """
        cb_str = json.dumps(callback_params).strip()
        return base64.b64encode(cb_str.encode()).decode()

    # 设置上传回调参数
    callback_params = {}
    callback_params['callbackUrl'] = args.callback_url  # 回调服务器地址
    callback_params['callbackBody'] = 'bucket=${bucket}&object=${object}&my_var_1=${x:my_var1}&my_var_2=${x:my_var2}'  # 回调请求体
    callback_params['callbackBodyType'] = 'application/x-www-form-urlencoded'  # 回调请求体类型
    encoded_callback = encode_callback(callback_params)  # 对回调参数进行编码

    # 添加回调相关字段到表单数据中
    field_dict['callback'] = encoded_callback
    field_dict['x:my_var1'] = 'value1'
    field_dict['x:my_var2'] = 'value2'

    # 生成随机边界字符串
    boundary = ''.join(random.choice('0123456789') for _ in range(11))
    # 发送POST请求
    body = build_post_body(field_dict, boundary)

    # 构建OSS服务的URL
    url = f"http://{bucket_name}.oss-{region}.aliyuncs.com"
    headers = {
        "Content-Type": f"multipart/form-data; boundary={boundary}",  # 设置请求头
    }

    # 发送POST请求
    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')}")

    # 打印响应内容
    print(f"response: {response.text}")

if __name__ == "__main__":
    main()

相关文档