本文介绍如何使用OSS Python SDK V2完成Agentic Bucket的创建与管理、泛域名CNAME的绑定与解绑,以及BucketSpace的创建与文件上传。
注意事项
使用Agentic Bucket相关功能,您需要安装V1.4.0及以上版本的alibabacloud-oss-v2。
本文示例代码以华东1(杭州)的地域ID
cn-hangzhou为例,默认使用外网Endpoint,如果您希望通过与OSS同地域的其他阿里云产品访问OSS,请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系,请参见地域和Endpoint。使用
alibabacloud_oss_v2.agentic模块(AgenticBucketClient、BucketSpaceClient)时,必须在配置中同时设置region和account_id。请求中的bucket参数为Bucket名称前缀,SDK会自动将其解析为完整名称:Agentic Bucket对应的完整名称为{prefix}-{account_id}-{region}-ab-apsr,BucketSpace对应的完整名称为{prefix}-{account_id}-{region}-bs-apsr。访问BucketSpace有两种方式,可按需选择:一是使用
BucketSpaceClient,请求中的bucket参数传入BucketSpace名称前缀,SDK自动将其解析为完整名称;二是使用BucketSpaceHelper和标准客户端oss.Client,先通过to_bucket_name方法将名称前缀转换为完整名称,再通过标准客户端访问BucketSpace,此方式下请求中的bucket参数须传入完整名称。使用BucketSpaceHelper时,同样必须在配置中设置region和account_id。CNAME相关操作(创建CnameToken、获取CnameToken、绑定CNAME、列举CNAME、删除CNAME)由标准客户端
oss.Client提供,请求中的bucket参数须传入Agentic Bucket的完整名称,即{prefix}-{account_id}-{region}-ab-apsr。泛域名CNAME相关请求中,
domain或cname参数须传入泛域名基础域(不含*.前缀),且基础域至少包含3段(例如example.com不合法,foo.example.com合法)。绑定泛域名CNAME前,需先为该基础域申请CnameToken并完成域名所有权验证;如果绑定证书,证书须为与*.基础域匹配的通配符证书。关于泛域名CNAME的完整规则,请参见PutCname。
示例代码
创建Agentic Bucket
以下代码用于创建Agentic Bucket。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:创建Agentic Bucket
parser = argparse.ArgumentParser(description="create agentic bucket sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及可选的Endpoint
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('--account_id', help='The account id.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性。AgenticBucketClient会根据二者
# 自动将Bucket名称前缀解析为完整名称:{prefix}-{account_id}-{region}-ab-apsr
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化Agentic Bucket客户端
client = oss_agentic.AgenticBucketClient(cfg)
# 创建Agentic Bucket,存储类型为Standard,数据容灾类型为LRS
result = client.create_agentic_bucket(oss_agentic.models.CreateAgenticBucketRequest(
bucket=args.bucket, # Bucket名称前缀
create_agentic_bucket_configuration=oss_agentic.models.CreateAgenticBucketConfiguration(
storage_class='Standard', # 存储类型
data_redundancy_type='LRS', # 数据容灾类型
),
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()查询Agentic Bucket信息
以下代码用于查询指定Agentic Bucket的信息。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:查询Agentic Bucket信息
parser = argparse.ArgumentParser(description="get agentic bucket sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及可选的Endpoint
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('--account_id', help='The account id.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化Agentic Bucket客户端
client = oss_agentic.AgenticBucketClient(cfg)
# 查询指定Agentic Bucket的信息。request.bucket为名称前缀
result = client.get_agentic_bucket(oss_agentic.models.GetAgenticBucketRequest(
bucket=args.bucket, # Bucket名称前缀
))
# 获取并打印Agentic Bucket的详细信息
info = result.agentic_bucket_info
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' name: {info.name},' # Agentic Bucket完整名称
f' region: {info.region},' # 所在地域
f' storage class: {info.storage_class},' # 存储类型
f' status: {info.status},' # 状态
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()列举Agentic Bucket
以下代码用于列举当前账号在指定地域下的所有Agentic Bucket。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:列举Agentic Bucket
parser = argparse.ArgumentParser(description="list agentic buckets sample")
# 定义命令行参数,包括必需的地域、账号ID以及可选的Endpoint
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--account_id', help='The account id.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化Agentic Bucket客户端
client = oss_agentic.AgenticBucketClient(cfg)
# 创建ListAgenticBuckets分页器
paginator = client.list_agentic_buckets_paginator()
# 逐页遍历当前账号在指定地域下的所有Agentic Bucket
for page in paginator.iter_page(oss_agentic.models.ListAgenticBucketsRequest()):
for o in page.agentic_buckets:
# 打印每个Agentic Bucket的完整名称和存储类型
print(f'agentic bucket: {o.name}, {o.storage_class}')
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()删除Agentic Bucket
以下代码用于删除指定的Agentic Bucket。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:删除Agentic Bucket
parser = argparse.ArgumentParser(description="delete agentic bucket sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及可选的Endpoint
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('--account_id', help='The account id.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化Agentic Bucket客户端
client = oss_agentic.AgenticBucketClient(cfg)
# 删除指定的Agentic Bucket。request.bucket为名称前缀
result = client.delete_agentic_bucket(oss_agentic.models.DeleteAgenticBucketRequest(
bucket=args.bucket, # Bucket名称前缀
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()创建CnameToken(泛域名)
以下代码用于为Agentic Bucket上的泛域名基础域申请CnameToken。
import argparse
import alibabacloud_oss_v2 as oss
# 创建命令行参数解析器,并描述脚本用途:为Agentic Bucket上的泛域名申请CnameToken
parser = argparse.ArgumentParser(description="create cname token sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及泛域名基础域
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('--account_id', help='The account id.', required=True)
parser.add_argument('--domain', help='The base domain name of the wildcard domain, without the "*." prefix.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化OSS客户端。CNAME相关操作由标准客户端提供,
# 请求中的bucket参数须传入Agentic Bucket的完整名称
client = oss.Client(cfg)
# Agentic Bucket的完整名称为{prefix}-{account_id}-{region}-ab-apsr
agentic_bucket = f'{args.bucket}-{args.account_id}-{args.region}-ab-apsr'
# 为泛域名基础域申请CnameToken。domain传入不含*.前缀的基础域,
# is_wild_card设置为true,表示为泛域名CNAME申请CnameToken
result = client.create_cname_token(oss.CreateCnameTokenRequest(
bucket=agentic_bucket, # Agentic Bucket完整名称
bucket_cname_configuration=oss.BucketCnameConfiguration(
cname=oss.Cname(
domain=args.domain, # 泛域名基础域,不含*.前缀
is_wild_card=True, # 标识泛域名CNAME
),
),
))
# 打印操作结果的状态码和其他相关信息,以便确认请求状态和CnameToken详情
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' cname: {result.cname_token.cname if hasattr(result.cname_token, "cname") else "Not set"},'
f' token: {result.cname_token.token if hasattr(result.cname_token, "token") else "Not set"},'
f' expire time: {result.cname_token.expire_time.strftime("%Y-%m-%dT%H:%M:%S.000Z") if hasattr(result.cname_token, "expire_time") else "Not set"},'
f' bucket: {result.cname_token.bucket if hasattr(result.cname_token, "bucket") else "Not set"},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()获取CnameToken(泛域名)
以下代码用于获取Agentic Bucket上泛域名CNAME的CnameToken。
import argparse
import alibabacloud_oss_v2 as oss
# 创建命令行参数解析器,并描述脚本用途:获取Agentic Bucket上泛域名的CnameToken
parser = argparse.ArgumentParser(description="get cname token sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及泛域名基础域
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('--account_id', help='The account id.', required=True)
parser.add_argument('--domain', help='The base domain name of the wildcard domain, without the "*." prefix.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化OSS客户端。CNAME相关操作由标准客户端提供,
# 请求中的bucket参数须传入Agentic Bucket的完整名称
client = oss.Client(cfg)
# Agentic Bucket的完整名称为{prefix}-{account_id}-{region}-ab-apsr
agentic_bucket = f'{args.bucket}-{args.account_id}-{args.region}-ab-apsr'
# 获取泛域名基础域的CnameToken。cname传入不含*.前缀的基础域,
# wildcard设置为true,表示获取泛域名CNAME的CnameToken
result = client.get_cname_token(oss.GetCnameTokenRequest(
bucket=agentic_bucket, # Agentic Bucket完整名称
cname=args.domain, # 泛域名基础域,不含*.前缀
wildcard=True, # 标识泛域名CNAME
))
# 打印操作结果的状态码和其他相关信息,以便确认请求状态和CnameToken详情
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' cname: {result.cname_token.cname if hasattr(result.cname_token, "cname") else "Not set"},'
f' token: {result.cname_token.token if hasattr(result.cname_token, "token") else "Not set"},'
f' expire time: {result.cname_token.expire_time.strftime("%Y-%m-%dT%H:%M:%S.000Z") if hasattr(result.cname_token, "expire_time") else "Not set"},'
f' bucket: {result.cname_token.bucket if hasattr(result.cname_token, "bucket") else "Not set"},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()绑定泛域名CNAME
以下代码用于为Agentic Bucket绑定泛域名CNAME。
import argparse
import alibabacloud_oss_v2 as oss
# 创建命令行参数解析器,并描述脚本用途:为Agentic Bucket绑定泛域名CNAME
parser = argparse.ArgumentParser(description="put cname sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及泛域名基础域
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('--account_id', help='The account id.', required=True)
parser.add_argument('--domain', help='The base domain name of the wildcard domain, without the "*." prefix.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化OSS客户端。CNAME相关操作由标准客户端提供,
# 请求中的bucket参数须传入Agentic Bucket的完整名称
client = oss.Client(cfg)
# Agentic Bucket的完整名称为{prefix}-{account_id}-{region}-ab-apsr
agentic_bucket = f'{args.bucket}-{args.account_id}-{args.region}-ab-apsr'
# SSL证书相关参数。泛域名CNAME须使用与*.基础域匹配的通配符证书
cert_id = '493****-cn-hangzhou' # 证书ID
certificate = '''-----BEGIN CERTIFICATE-----
MIIDhDCCAmwCCQCFs8ixARsyrDANBgkqhkiG9w0BAQsFADCBgzELMAkGA1UEBhMC
**** -----END CERTIFICATE-----''' # SSL证书内容
private_key = '''-----BEGIN CERTIFICATE-----
MIIDhDCCAmwCCQCFs8ixARsyrDANBgkqhkiG9w0BAQsFADCBgzELMAkGA1UEBhMC
**** -----END CERTIFICATE-----''' # 私钥内容
# 为Agentic Bucket绑定泛域名CNAME。domain传入不含*.前缀的基础域,
# is_wild_card设置为true,表示绑定泛域名CNAME
result = client.put_cname(oss.PutCnameRequest(
bucket=agentic_bucket, # Agentic Bucket完整名称
bucket_cname_configuration=oss.BucketCnameConfiguration(
cname=oss.Cname(
domain=args.domain, # 泛域名基础域,不含*.前缀
is_wild_card=True, # 标识泛域名CNAME
certificate_configuration=oss.CertificateConfiguration(
certificate=certificate, # SSL证书内容
private_key=private_key, # 私钥内容
cert_id=cert_id, # 证书ID
),
),
),
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()列举CNAME记录
以下代码用于列举Agentic Bucket下的CNAME记录。
import argparse
import alibabacloud_oss_v2 as oss
# 创建命令行参数解析器,并描述脚本用途:列举Agentic Bucket下的CNAME记录
parser = argparse.ArgumentParser(description="list cname sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及可选的Endpoint
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('--account_id', help='The account id.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化OSS客户端。CNAME相关操作由标准客户端提供,
# 请求中的bucket参数须传入Agentic Bucket的完整名称
client = oss.Client(cfg)
# Agentic Bucket的完整名称为{prefix}-{account_id}-{region}-ab-apsr
agentic_bucket = f'{args.bucket}-{args.account_id}-{args.region}-ab-apsr'
# 发送请求以列举指定Agentic Bucket下的CNAME记录
result = client.list_cname(oss.ListCnameRequest(
bucket=agentic_bucket, # Agentic Bucket完整名称
))
# 打印操作结果的状态码、请求ID和其他相关信息,以便确认请求状态和CNAME列表详情
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' bucket: {result.bucket},' # 存储空间名
f' owner: {result.owner},' # 存储空间所有者
)
# 如果存在CNAME记录,则遍历并打印每条CNAME记录的详细信息
if result.cnames:
for r in result.cnames:
# is_wild_card为true表示该条记录为泛域名CNAME
print(f'domain: {r.domain}, '
f'last modified: {r.last_modified}, '
f'is wild card: {r.is_wild_card}, '
f'status: {r.status}'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()删除泛域名CNAME
以下代码用于删除Agentic Bucket上的泛域名CNAME。
import argparse
import alibabacloud_oss_v2 as oss
# 创建命令行参数解析器,并描述脚本用途:删除Agentic Bucket上的泛域名CNAME
parser = argparse.ArgumentParser(description="delete cname sample")
# 定义命令行参数,包括必需的地域、Bucket名称前缀、账号ID以及泛域名基础域
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('--account_id', help='The account id.', required=True)
parser.add_argument('--domain', help='The base domain name of the wildcard domain, without the "*." prefix.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置初始化OSS客户端。CNAME相关操作由标准客户端提供,
# 请求中的bucket参数须传入Agentic Bucket的完整名称
client = oss.Client(cfg)
# Agentic Bucket的完整名称为{prefix}-{account_id}-{region}-ab-apsr
agentic_bucket = f'{args.bucket}-{args.account_id}-{args.region}-ab-apsr'
# 删除Agentic Bucket上的泛域名CNAME。domain传入不含*.前缀的基础域,
# is_wild_card设置为true,表示删除泛域名CNAME
result = client.delete_cname(oss.DeleteCnameRequest(
bucket=agentic_bucket, # Agentic Bucket完整名称
bucket_cname_configuration=oss.BucketCnameConfiguration(
cname=oss.Cname(
domain=args.domain, # 泛域名基础域,不含*.前缀
is_wild_card=True, # 标识泛域名CNAME
),
),
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()创建BucketSpace
以下代码用于在指定的Agentic Bucket下创建BucketSpace。SDK提供两种访问BucketSpace的方式,以下示例分别演示,您可按需选择。
方式一:使用BucketSpaceClient,请求中的bucket参数传入BucketSpace名称前缀。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:创建BucketSpace
parser = argparse.ArgumentParser(description="bucket space put bucket sample")
# 定义命令行参数,包括必需的地域、BucketSpace名称前缀、账号ID、
# 所属Agentic Bucket名称前缀以及可选的Endpoint
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The bucket space prefix name.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--account_id', help='The account id.', required=True)
parser.add_argument('--agentic_bucket', help='The agentic bucket that the bucket space belongs to.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 创建BucketSpace客户端。该客户端复用标准OSS Bucket/Object操作,
# 并自动将BucketSpace名称前缀解析为完整名称{prefix}-{account_id}-{region}-bs-apsr
client = oss_agentic.BucketSpaceClient.create(cfg)
# 在指定的Agentic Bucket下创建BucketSpace。PutBucketRequest的bucket参数
# 为BucketSpace名称前缀,agentic_bucket参数为所属Agentic Bucket的完整名称
result = client.put_bucket(oss.models.PutBucketRequest(
bucket=args.bucket, # BucketSpace名称前缀
agentic_bucket=f'{args.agentic_bucket}-{args.account_id}-{args.region}-ab-apsr', # AgenticBucket完整名称
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()方式二:使用BucketSpaceHelper和标准客户端oss.Client,请求中的bucket参数传入转换后的BucketSpace完整名称。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:使用BucketSpaceHelper创建BucketSpace
parser = argparse.ArgumentParser(description="bucket space put bucket sample with helper")
# 定义命令行参数,与方式一相同,包括必需的地域、BucketSpace名称前缀、账号ID、
# 所属Agentic Bucket名称前缀以及可选的Endpoint
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The bucket space prefix name.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--account_id', help='The account id.', required=True)
parser.add_argument('--agentic_bucket', help='The agentic bucket that the bucket space belongs to.', required=True)
def main():
# 解析命令行参数,获取用户输入的值
args = parser.parse_args()
# 从环境变量中加载访问凭证信息,用于身份验证
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK默认配置创建配置对象,并设置认证提供者
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 创建标准OSS客户端
client = oss.Client(cfg)
# 创建BucketSpaceHelper。该Helper根据配置中的账号ID和地域,
# 将BucketSpace名称前缀转换为完整名称{prefix}-{account_id}-{region}-bs-apsr
helper = oss_agentic.BucketSpaceHelper(cfg)
# 在指定的Agentic Bucket下创建BucketSpace。PutBucketRequest的bucket参数
# 为helper.to_bucket_name转换后的BucketSpace完整名称,agentic_bucket参数为所属Agentic Bucket的完整名称
result = client.put_bucket(oss.models.PutBucketRequest(
bucket=helper.to_bucket_name(args.bucket), # BucketSpace完整名称
agentic_bucket=f'{args.agentic_bucket}-{args.account_id}-{args.region}-ab-apsr', # Agentic Bucket完整名称
))
# 打印操作结果的状态码和请求ID,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()向BucketSpace上传文件
以下代码用于向指定的BucketSpace上传文件(Object)。以下示例分别演示两种访问方式。
方式一:使用BucketSpaceClient,请求中的bucket参数传入BucketSpace名称前缀。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:向BucketSpace上传文件
parser = argparse.ArgumentParser(description="bucket space put object sample")
# 定义命令行参数,包括必需的地域、BucketSpace名称前缀、账号ID、
# 文件名称以及可选的Endpoint
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The bucket space prefix name.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--account_id', help='The account id.', required=True)
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
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 创建BucketSpace客户端。该客户端复用标准OSS Bucket/Object操作,
# 并自动将BucketSpace名称前缀解析为完整名称{prefix}-{account_id}-{region}-bs-apsr
client = oss_agentic.BucketSpaceClient.create(cfg)
# 向指定的BucketSpace上传文件
result = client.put_object(oss.models.PutObjectRequest(
bucket=args.bucket, # BucketSpace名称前缀
key=args.key, # 文件名称
body=b'hello agentic bucket space', # 文件内容
))
# 打印操作结果的状态码、请求ID以及文件的ETag,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' etag: {result.etag},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()方式二:使用BucketSpaceHelper和标准客户端oss.Client,请求中的bucket参数传入转换后的BucketSpace完整名称。
import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.agentic as oss_agentic
# 创建命令行参数解析器,并描述脚本用途:使用BucketSpaceHelper向BucketSpace上传文件
parser = argparse.ArgumentParser(description="bucket space put object sample with helper")
# 定义命令行参数,与方式一相同,包括必需的地域、BucketSpace名称前缀、账号ID、
# 文件名称以及可选的Endpoint
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The bucket space prefix name.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--account_id', help='The account id.', required=True)
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
# 设置配置对象的地域和账号ID属性
cfg.region = args.region
cfg.account_id = args.account_id
# 如果提供了自定义Endpoint,则更新配置对象中的endpoint属性
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 创建标准OSS客户端
client = oss.Client(cfg)
# 创建BucketSpaceHelper,用于将BucketSpace名称前缀转换为完整名称
# {prefix}-{account_id}-{region}-bs-apsr
helper = oss_agentic.BucketSpaceHelper(cfg)
# 向指定的BucketSpace上传文件。PutObjectRequest的bucket参数
# 为helper.to_bucket_name转换后的BucketSpace完整名称
result = client.put_object(oss.models.PutObjectRequest(
bucket=helper.to_bucket_name(args.bucket), # BucketSpace完整名称
key=args.key, # 文件名称
body=b'hello agentic bucket space', # 文件内容
))
# 打印操作结果的状态码、请求ID以及文件的ETag,以便确认请求状态
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' etag: {result.etag},'
)
# 当此脚本被直接执行时,调用main函数开始处理逻辑
if __name__ == "__main__":
main()相关文档
关于创建Agentic Bucket的完整示例代码,请参见create_agentic_bucket.py。
关于查询Agentic Bucket信息的完整示例代码,请参见get_agentic_bucket.py。
关于列举Agentic Bucket的完整示例代码,请参见list_agentic_buckets.py。
关于删除Agentic Bucket的完整示例代码,请参见delete_agentic_bucket.py。
关于创建域名所有权验证所需CnameToken的完整示例代码,请参见create_cname_token.py。
关于获取CnameToken的完整示例代码,请参见get_cname_token.py。
关于添加CNAME记录的完整示例代码,请参见put_cname.py。
关于查看CNAME记录的完整示例代码,请参见list_cname.py。
关于删除CNAME记录的完整示例代码,请参见delete_cname.py。
关于创建BucketSpace的完整示例代码,请参见bucket_space_put_bucket.py。
关于向BucketSpace上传文件的完整示例代码,请参见bucket_space_put_object.py。
关于CNAME相关接口的详细参数说明,请参见CreateCnameToken、PutCname、ListCname、GetCnameToken和DeleteCname。