如何通过Python SDK V2上传文件_对象存储(OSS)-阿里云帮助中心

本文介绍如何通过简单上传方法将单个文件快速上传到OSS，此方法操作简便，适合快速将文件上传到云端存储。

注意事项

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

权限说明

阿里云账号默认拥有全部权限。阿里云账号下的RAM用户或RAM角色默认没有任何权限，需要阿里云账号或账号管理员通过RAM Policy或Bucket Policy授予操作权限。

API	Action	说明
PutObject	`oss:PutObject`	上传Object。
	`oss:PutObjectTagging`	上传Object时，如果通过x-oss-tagging指定Object的标签，则需要此操作的权限。
	`kms:GenerateDataKey`	上传Object时，如果Object的元数据包含X-Oss-Server-Side-Encryption: KMS，则需要这两个操作的权限。
	`kms:Decrypt`

方法定义

put_object(request: PutObjectRequest, **kwargs) → PutObjectResult

请求参数列表

参数名	类型	说明
request	PutObjectRequest	设置请求参数，例如设置对象的访问控制方式（Acl）、禁止覆盖写（ForbidOverwrite）、自定义元数据（Metadata）等，具体请参见PutObjectRequest

返回值列表

类型	说明
PutObjectResult	返回值，具体请参见PutObjectResult

关于简单上传方法的完整定义，请参见put_object。

上传本地文件

上传文件（Object）时，如果存储空间（Bucket）中已存在同名文件且用户对该文件有访问权限，则新添加的文件将覆盖原有文件。

上传文件时涉及填写的公共参数如下：

参数	说明
bucket_name	Bucket名称。 Bucket名称的命名规范如下：只能包括小写字母、数字和短划线（-）。必须以小写字母或者数字开头和结尾。长度必须在3~63字符之间。
object_name	Object完整路径。Object完整路径中不能包含Bucket名称。 Object命名规范如下：使用UTF-8编码。长度必须在1~1023字符之间。不能以正斜线（/）或者反斜线（\）开头。

使用put_object_from_file方法直接将本地文件上传到目标存储空间。

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put object from file 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)

# 添加命令行参数 --file_path，表示要上传的本地文件路径，必需参数
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置中的区域信息
    cfg.region = args.region

    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 执行上传对象的请求，直接从文件上传
    # 指定存储空间名称、对象名称和本地文件路径
    result = client.put_object_from_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 存储空间名称
            key=args.key         # 对象名称
        ),
        args.file_path          # 本地文件路径
    )

    # 输出请求的结果信息，包括状态码、请求ID、内容MD5、ETag、CRC64校验码、版本ID和服务器响应时间
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
    )

# 脚本入口，当文件被直接运行时调用main函数
if __name__ == "__main__":
    main()

使用put_object方法上传本地文件，必须使用'rb'模式打开文件，以确保上传的是原始字节流而非文本内容，避免出现CRC校验失败。

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put object from file 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():
    # 解析命令行参数
    args = parser.parse_args()

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 设置配置中的区域信息
    cfg.region = args.region

    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 执行上传对象的请求，直接从本地文件上传
    # 指定存储空间名称、对象名称和本地文件路径
    with open('your-test-file.md', 'rb') as f:
    result = client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 存储空间名称
            key=args.key,        # 对象名称
            body=f.read()        # 读取文件内容
        )
    )

    # 输出请求的结果信息，包括状态码、请求ID、内容MD5、ETag、CRC64校验码、版本ID和服务器响应时间
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
    )

# 脚本入口，当文件被直接运行时调用main函数
if __name__ == "__main__":
    main()

常见使用场景

上传字符串

您可以使用以下代码将字符串上传至目标存储空间。

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put 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():
    args = parser.parse_args()  # 解析命令行参数

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    # 设置配置中的区域信息
    cfg.region = args.region
    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 定义要上传的字符串内容
    text_string = "Hello, OSS!"
    data = text_string.encode('utf-8')  # 将字符串编码为UTF-8字节串

    # 执行上传对象的请求，指定存储空间名称、对象名称和数据内容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
    ))

    # 输出请求的结果状态码、请求ID、内容MD5、ETag、CRC64校验码和版本ID，用于检查请求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 脚本入口，当文件被直接运行时调用main函数

上传字节数组

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put 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():
    args = parser.parse_args()  # 解析命令行参数

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    # 设置配置中的区域信息
    cfg.region = args.region
    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 定义要上传的数据内容
    data = b'hello world'

    # 执行上传对象的请求，指定存储空间名称、对象名称和数据内容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
    ))

    # 输出请求的结果状态码、请求ID、内容MD5、ETag、CRC64校验码和版本ID，用于检查请求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 脚本入口，当文件被直接运行时调用main函数

上传网络流

您可以使用以下代码将网络流上传至目标存储空间。

import argparse
import requests
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put 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():
    args = parser.parse_args()  # 解析命令行参数

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    # 设置配置中的区域信息
    cfg.region = args.region
    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 发送HTTP GET请求，获取响应内容
    response = requests.get('http://www.aliyun.com')

    # 执行上传对象的请求，指定存储空间名称、对象名称和数据内容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=response.content,
    ))

    # 输出请求的结果状态码、请求ID、内容MD5、ETag、CRC64校验码和版本ID，用于检查请求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 脚本入口，当文件被直接运行时调用main函数

上传文件并设置回调

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

import base64
import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put 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('--call_back_url', help='Callback server address.', required=True)


def main():

    args = parser.parse_args()

    # 从环境变量中加载凭证信息
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 配置 SDK 客户端
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 创建 OSS 客户端
    client = oss.Client(cfg)

    # 要上传的内容（字符串）
    data = 'hello world'

    # 构造回调参数（callback）：指定回调地址和回调请求体，使用 Base64 编码
    callback=base64.b64encode(str('{\"callbackUrl\":\"' + args.call_back_url + '\",\"callbackBody\":\"bucket=${bucket}&object=${object}&my_var_1=${x:var1}&my_var_2=${x:var2}\"}').encode()).decode(),
    # 构造自定义变量（callback-var），使用 Base64 编码
    callback_var=base64.b64encode('{\"x:var1\":\"value1\",\"x:var2\":\"value2\"}'.encode()).decode(),

    # 发起上传请求，并携带回调参数
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
        callback=callback,
        callback_var=callback_var,
    ))
    # 打印返回结果（包括状态码、请求 ID 等）
    print(vars(result))


if __name__ == "__main__":
    main()

上传文件显示进度条

以下示例代码展示了在上传本地文件时，使用进度条查看上传进度。

import argparse
import alibabacloud_oss_v2 as oss

# 创建命令行参数解析器
parser = argparse.ArgumentParser(description="put 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():
    args = parser.parse_args()  # 解析命令行参数

    # 从环境变量中加载凭证信息，用于身份验证
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 加载SDK的默认配置，并设置凭证提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    # 设置配置中的区域信息
    cfg.region = args.region
    # 如果提供了endpoint参数，则设置配置中的endpoint
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用配置好的信息创建OSS客户端
    client = oss.Client(cfg)

    # 定义一个字典变量 progress_state 用于保存上传进度状态，初始值为 0
    progress_state = {'saved': 0}
    def _progress_fn(n, written, total):
        # 使用字典存储累计写入的字节数，避免使用 global 变量
        progress_state['saved'] += n

        # 计算当前上传百分比，将已写入字节数与总字节数进行除法运算后取整
        rate = int(100 * (float(written) / float(total)))

        # 打印当前上传进度，\r 表示回到行首，实现命令行中实时刷新效果
        # end='' 表示不换行，使下一次打印覆盖当前行
        print(f'\r上传进度：{rate}% ', end='')

    # 执行上传对象的请求，指定存储空间名称、对象名称和数据内容
    result = client.put_object_from_file(oss.PutObjectRequest(
            bucket=args.bucket,
            key=args.key,
            progress_fn=_progress_fn,
        ),
        "/local/dir/example", # 指定本地文件路径
    )

    # 输出请求的结果状态码、请求ID、内容MD5、ETag、CRC64校验码和版本ID，用于检查请求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 脚本入口，当文件被直接运行时调用main函数

简单上传（Python SDK V2）