Agentic Bucket(Python SDK V2)

更新时间:
复制 MD 格式

本文介绍如何使用OSS Python SDK V2完成Agentic Bucket的创建与管理、泛域名CNAME的绑定与解绑,以及BucketSpace的创建与文件上传。

注意事项

  • 使用Agentic Bucket相关功能,您需要安装V1.4.0及以上版本的alibabacloud-oss-v2。

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

  • 使用alibabacloud_oss_v2.agentic模块(AgenticBucketClient、BucketSpaceClient)时,必须在配置中同时设置regionaccount_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时,同样必须在配置中设置regionaccount_id

  • CNAME相关操作(创建CnameToken、获取CnameToken、绑定CNAME、列举CNAME、删除CNAME)由标准客户端oss.Client提供,请求中的bucket参数须传入Agentic Bucket的完整名称,即{prefix}-{account_id}-{region}-ab-apsr

  • 泛域名CNAME相关请求中,domaincname参数须传入泛域名基础域(不含*.前缀),且基础域至少包含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上泛域名CNAMECnameToken。

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()