注意事项
本文示例代码以华东1(杭州)的地域IDcn-hangzhou
为例,默认使用外网Endpoint,如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系,请参见OSS地域和访问域名。
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 |
返回值列表
关于简单上传方法的完整定义,请参见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函数