OSS Python SDK V2

快速接入

环境准备

要求 Python 3.8 及以上版本。

可以通过 Python -version 命令查看 Python 版本。如果当前环境没有 Python 或版本低于 Python 3.8，请下载并安装Python。

安装SDK

• 执行以下命令安装OSS Python SDK V2代码包。请根据需求选择合适的OSS Python SDK V2版本，推荐使用最新的版本，确保本文中的代码示例可以正常运行。关于版本功能的更多信息，请参见Releases。

  pip install alibabacloud-oss-v2

• 使用以下代码引入OSS Python SDK V2包。

  import alibabacloud_oss_v2 as oss

初始化客户端

运行示例代码前，请将代码中的 <region-id>等占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

def main():
    """
    Python SDK V2 客户端初始化配置说明：

    1. 签名版本：Python SDK V2 默认使用 V4 签名，提供更高的安全性
    2. Region配置：初始化 Client 时，必须指定阿里云 Region ID 作为请求地域标识
    3. Endpoint配置：
       - 可通过Endpoint参数自定义服务请求的访问域名
       - 当不指定 Endpoint 时，将根据 Region 自动构造公网访问域名
    4. 协议配置：
       - SDK 默认使用 HTTPS 协议构造访问域名
       - 如需使用 HTTP 协议，在指定域名时明确指定
    """

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

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

    # 方式一：只填写Region（推荐）
    # 必须指定Region ID，SDK会根据Region自动构造HTTPS访问域名
    cfg.region = '<region-id>' 

    # # 方式二：同时填写Region和Endpoint
    # # 必须指定Region ID
    # cfg.region = '<region-id>'
    # # 填写Bucket所在地域对应的外网Endpoint
    # cfg.endpoint = '<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="Your Bucket Name",
        key="Your Object Key",
        body=data,
    ))

    # 输出请求的结果状态码、请求ID、ETag，用于检查请求是否成功
    print(f'status code: {result.status_code}\n'
          f'request id: {result.request_id}\n'
          f'etag: {result.etag}'
    )

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

import asyncio
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.aio as oss_aio

async def main():
    """
    Python SDK V2 异步客户端初始化配置说明：

    1. 签名版本：Python SDK V2 默认使用 V4 签名，提供更高的安全性

    2. Region配置：初始化 AsyncClient 时，必须指定阿里云 Region ID 作为请求地域标识

    3. Endpoint配置：
       - 可通过Endpoint参数自定义服务请求的访问域名
       - 当不指定 Endpoint 时，将根据 Region 自动构造公网访问域名

    4. 协议配置：
       - SDK 默认使用 HTTPS 协议构造访问域名
       - 如需使用 HTTP 协议，在指定域名时明确指定
    
    5. 异步特性：
       - 导入异步模块：import alibabacloud_oss_v2.aio as oss_aio
       - 创建异步客户端：oss_aio.AsyncClient(cfg)
       - 所有操作需要使用 await 关键字
       - 需要在 finally 块中调用 await client.close() 关闭连接
       - 需要安装 aiohttp 依赖：pip install aiohttp
    """

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

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

    # 方式一：只填写Region（推荐）
    # 必须指定Region ID，SDK会根据Region自动构造HTTPS访问域名
    cfg.region = '<region-id>'

    # # 方式二：同时填写Region和Endpoint
    # # 必须指定Region ID
    # cfg.region = '<region-id>'
    # # 填写Bucket所在地域对应的外网Endpoint
    # cfg.endpoint = '<endpoint>'

    # 使用配置好的信息创建OSS异步客户端
    client = oss_aio.AsyncClient(cfg)

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

        # 执行异步上传对象的请求，指定存储空间名称、对象名称和数据内容
        # 注意：使用 await 关键字等待异步操作完成
        result = await client.put_object(
            oss.PutObjectRequest(
                bucket="Your Bucket Name",
                key="Your Object Key",
                body=data,
            )
        )

        # 输出请求的结果状态码、请求ID、ETag，用于检查请求是否成功
        print(f'status code: {result.status_code}\n'
              f'request id: {result.request_id}\n'
              f'etag: {result.etag}'
        )

    except Exception as e:
        print(f'上传失败: {e}')

    finally:
        # 关闭异步客户端连接（重要：避免资源泄漏）
        await client.close()

# 当此脚本被直接运行时，调用main函数
if __name__ == "__main__":
    # 使用 asyncio.run() 运行异步主函数
    asyncio.run(main())

运行后将会输出上传文件成功的结果：

status code: 200
request id: 6875F95738B0ED3030F086A0
etag: "56AAD346F0899BFE8BDD02C06BBE511E"

客户端配置

使用自定义域名

使用OSS默认域名访问时，可能会出现文件禁止访问、文件无法预览等问题；通过通过自定义域名访问OSS，不仅支持浏览器直接预览文件，还可结合CDN加速分发。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

    # 请填写您的自定义域名。例如www.example-***.com
    cfg.endpoint = 'https://www.example-***.com'

    # 请注意，设置true开启CNAME选项，否则无法使用自定义域名
    cfg.use_cname = True

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

    # 使用创建好的client执行后续操作...

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

超时控制

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

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

    # 加载默认的配置信息
    cfg = oss.config.load_default()

    # 设置建立连接的超时时间, 默认值 10秒
    cfg.connect_timeout = 30

    # 设置应用读写数据的超时时间, 默认值 20秒
    cfg.readwrite_timeout = 30

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

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

    # 使用创建好的client执行后续操作...

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

重试策略

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

def main():
    """
    SDK 重试策略配置说明：

    默认重试策略：
    当没有配置重试策略时，SDK 使用 StandardRetryer() 作为客户端的默认实现，其默认配置如下：
    - max_attempts：设置最大尝试次数。默认为3次
    - max_backoff：设置最大退避时间（单位：秒）。默认为20秒
    - base_delay：设置基础延迟时间（单位：秒）。默认为0.2秒
    - backoff_delayer：设置退避算法。默认使用FullJitter退避算法
      计算公式为：[0.0, 1.0) * min(2^attempts * baseDelay, maxBackoff)
    - error_retryables：可重试的错误类型，请参阅https://gosspublic.alicdn.com/sdk-doc/alibabacloud-oss-python-sdk-v2/latest/_modules/alibabacloud_oss_v2/retry/error_retryable.html

    当发生可重试错误时，将使用其提供的配置来延迟并随后重试该请求。
    请求的总体延迟会随着重试次数而增加，如果默认配置不满足您的场景需求时，
    可配置重试参数或者修改重试实现。
    """

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

    # 加载默认的配置信息
    cfg = oss.config.load_default()

    # 配置重试策略示例：

    # 1. 自定义最大重试次数（默认为3次，这里设置为5次）
    cfg.retryer = oss.retry.StandardRetryer(max_attempts=5)

    # 2. 自定义退避延迟时间
    # 调整基础延迟时间 base_delay 为0.5秒（默认0.2秒），最大退避时间 max_backoff 为25秒（默认20秒）
    # cfg.retryer = oss.retry.StandardRetryer(max_backoff=25, base_delay=0.5)

    # 3. 自定义退避算法
    # 使用固定时间退避算法替代默认的FullJitter算法，每次延迟2秒
    # cfg.retryer = oss.retry.StandardRetryer(backoff_delayer=oss.retry.FixedDelayBackoff(2))

    # 4. 禁用重试策略
    # 如需禁用所有重试尝试时，可以使用retry.NopRetryer实现
    # cfg.retryer = oss.retry.NopRetryer()

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

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

    # 使用创建好的client执行后续操作...

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

设置连接池大小

在http_client中指定max_connections参数即可配置连接池大小。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# OSS Python SDK V2 连接池配置示例

import alibabacloud_oss_v2 as oss

def main():

    # 从环境变量中加载访问凭证（需要设置OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

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

    # 设置OSS服务地域
    config.region = "<region-id>"

    # ============ 自定义HTTPClient配置 ============
    # 创建HTTP客户端配置参数字典
    http_client_config = {}
    
    # 设置连接池大小为100，默认为20
    http_client_config["max_connections"] = 100
    
    # 使用自定义配置创建HTTP客户端
    http_client = oss.transport.RequestsHttpClient(**http_client_config)
    
    # 将自定义的HTTP客户端设置到配置对象中
    config.http_client = http_client
    # ============================================
    
    # 初始化OSS客户端实例
    client = oss.Client(config)

    # 配置Bucket和文件信息
    bucket = "example-bucket"             # Bucket名称
    key = "dest.jpg"                      # OSS中的文件路径
    file_path = "dest.jpg"                # 本地保存路径

    # 将OSS中的文件下载到本地指定路径
    client.get_object_to_file(oss.GetObjectRequest(bucket, key), file_path)
    print(f"文件下载完成: {key} -> {file_path}")
    print(f"连接池大小已设置为: {http_client_config['max_connections']}")

if __name__ == "__main__":
    # 程序入口点
    try:
        main()
    except Exception as e:
        print(f"运行过程中发生错误: {e}")
        raise

HTTP/HTTPS 协议

使用 cfg.disable_ssl = True 设置不使用HTTPS协议。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

    # 设置不使用https请求
    cfg.disable_ssl = True

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

    # 使用创建好的client执行后续操作...

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

代理服务器

企业安全策略通常限制直接访问公网，配置代理服务器访问 OSS。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

    # 设置user_agent
    cfg.user_agent = 'aliyun-sdk-python'

    # 设置代理服务器
    cfg.proxy_host = 'http://user:passswd@proxy.example-***.com'

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

    # 使用创建好的client执行后续操作...

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

使用内网域名

使用内网域名访问同地域的OSS资源，可以降低流量成本并提高访问速度。

运行示例代码前，请将代码中的 <region-id>等占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 方式一： 填写Region并设置use_internal_endpoint为true
    # 填写Bucket所在地域
    cfg.region = '<region-id>'
    cfg.use_internal_endpoint = True

    # # 方式二： 直接填写Region和Endpoint
    # # 填写Bucket所在地域
    # cfg.region = '<region-id>'
    # # 填写Bucket所在地域对应的内网Endpoint
    # cfg.endpoint = '<endpoint>'

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

    # 使用创建好的client执行后续操作...

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

使用传输加速域名

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 方式一： 填写Region并设置use_accelerate_endpoint为true
    # 填写Bucket所在地域
    cfg.region = '<region-id>'
    cfg.use_accelerate_endpoint = True

    # # 方式二： 直接填写Region和传输加速Endpoint
    # # 填写Bucket所在地域
    # cfg.region = '<region-id>'
    # # 填写Bucket所在地域对应的传输加速Endpoint
    # cfg.endpoint = 'https://oss-accelerate.aliyuncs.com'

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

    # 使用创建好的client执行后续操作...

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

使用专有域

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

    # 请填写您的专有域。例如：https://service.corp.example.com
    cfg.endpoint = 'https://service.corp.example.com'

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

    # 使用创建好的client执行后续操作...

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

使用金融云域名

以下是使用金融云域名配置OSSClient的示例代码。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Region和Endpoint
    # 填写Bucket所在地域。以华东1 金融云为例，Region填写为cn-hangzhou-finance
    cfg.region = 'cn-hangzhou-finance'
    # 填写Bucket所在地域对应的内网Endpoint。以华东1 金融云为例，Endpoint填写为'https://oss-cn-hzjbp-a-internal.aliyuncs.com',
    # 如需指定为http协议，请在指定域名时填写为'http://oss-cn-hzjbp-a-internal.aliyuncs.com'
    cfg.endpoint = 'https://oss-cn-hzjbp-a-internal.aliyuncs.com'

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

    # 使用创建好的client执行后续操作...

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

使用政务云域名

以下是使用政务云域名配置OSSClient的示例代码。

import alibabacloud_oss_v2 as oss

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

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

    # 填写Region和Endpoint
    # 填写Bucket所在地域。以华北2 阿里政务云1为例，Region填写为cn-north-2-gov-1
    cfg.region = 'cn-north-2-gov-1'
    # 填写Bucket所在地域对应的内网Endpoint。以华北2 阿里政务云1为例，Endpoint填写为'https://oss-cn-north-2-gov-1-internal.aliyuncs.com',
    # 如需指定为http协议，请在指定域名时填写为'http://oss-cn-north-2-gov-1-internal.aliyuncs.com'
    cfg.endpoint = 'https://oss-cn-north-2-gov-1-internal.aliyuncs.com'

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

    # 使用创建好的client执行后续操作...

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

自定义HTTPClient

当常用配置参数无法满足场景需求时，您可以使用cfg.http_client替换默认的 HTTP 客户端。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

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

    # 加载默认的配置信息
    cfg = oss.config.load_default()

    # 设置HTTP客户端的参数
    kwargs: Dict[str, Any] = {}

    # 设置session
    # kwargs["session"] = requests.Session()

    # 设置adapter
    # kwargs["adapter"] = HTTPAdapter()

    # 是否跳过证书检查，默认不跳过
    # kwargs["insecure_skip_verify"] = False

    # 是否打开启HTTP重定向，默认不启用
    # kwargs["enabled_redirect"] = False

    # 设置代理服务器
    # kwargs["proxy_host"] = config.proxy_host

    # 设置块大小
    # kwargs["block_size"] = 16 * 1024

    # 连接超时, 默认值 10秒
    kwargs["connect_timeout"] = 30

    # 应用读写数据的超时时间, 默认值 20秒
    kwargs["readwrite_timeout"] = 30

    # 最大连接数，默认值 20
    kwargs["max_connections"] = 1024

    # 创建HTTP客户端，并传入HTTP客户端参数
    cfg.http_client = oss.transport.RequestsHttpClient(**kwargs)

    # 设置凭证提供者
    cfg.credentials_provider = credentials_provider

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

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

    # 使用创建好的client执行后续操作...

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

访问凭证配置

OSS 提供多种凭证初始化方式。请根据您的认证和授权需求选择合适的初始化方式。

使用RAM用户的AK

如果您的应用程序部署运行在安全、稳定且不易受外部攻击的环境中，需要长期访问您的OSS，且不能频繁轮转凭证时，您可以使用阿里云主账号或RAM用户的AK（Access Key ID、Access Key Secret）初始化凭证提供者。需要注意的是，该方式需要您手动维护一个AK，存在安全性风险和维护复杂度增加的风险。

import alibabacloud_oss_v2 as oss

def main():
    # 创建静态凭证提供者，显式设置访问密钥和密钥密码，请替换为您的RAM用户的AccessKey ID和AccessKey Secret
    credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id="RAM AccessKey ID",
        access_key_secret="RAM AccessKey Secret"
    )

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

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

    # 使用创建好的client执行后续操作...

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

使用STS临时访问凭证

如果您的应用程序需要临时访问OSS，您可以使用通过STS服务获取的临时身份凭证（Access Key ID、Access Key Secret和Security Token）初始化凭证提供者。需要注意的是，该方式需要您手动维护一个STS Token，存在安全性风险和维护复杂度增加的风险。此外，如果您需要多次临时访问OSS，您需要手动刷新STS Token。

import alibabacloud_oss_v2 as oss

def main():
    # 填写获取的临时访问密钥AccessKey ID和AccessKey Secret，非阿里云账号AccessKey ID和AccessKey Secret。
    # 请注意区分STS服务获取的Access Key ID是以STS开头，。。
    sts_access_key_id = 'STS.****************'
    sts_access_key_secret = 'yourAccessKeySecret'
    # 填写获取的STS安全令牌（SecurityToken）。
    sts_security_token = 'yourSecurityToken'

    # 创建静态凭证提供者，显式设置临时访问密钥AccessKey ID和AccessKey Secret，以及STS安全令牌
    credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id=sts_access_key_id,
        access_key_secret=sts_access_key_secret,
        security_token=sts_security_token,
    )

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

    # 填写Bucket所在地域
    cfg.region = '<region-id>'

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

    # 使用创建好的client执行后续操作...

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

使用RAMRoleARN

如果您的应用程序需要授权访问OSS，例如跨阿里云账号访问OSS，您可以使用RAMRoleARN初始化凭证提供者。该方式底层实现是STS Token。通过指定RAM角色的ARN（Alibabacloud Resource Name），Credentials工具会前往STS服务获取STS Token，并在会话到期前调用AssumeRole接口申请新的STS Token。此外，您还可以通过为policy赋值来限制RAM角色到一个更小的权限集合。

1. 添加alibabacloud_credentials依赖。

   pip install alibabacloud_credentials

2. 配置AK和RAMRoleARN作为访问凭证。

   运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

   # -*- coding: utf-8 -*-
   import os
   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   import alibabacloud_oss_v2 as oss
   
   def main():
       config = Config(
           # 从环境变量中获取RAM用户的访问密钥（AccessKey ID和AccessKey Secret）
           access_key_id=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
           access_key_secret=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
           type='ram_role_arn',
           # 要扮演的RAM角色ARN，示例值：acs:ram::123456789012****:role/adminrole，可以通过环境变量ALIBABA_CLOUD_ROLE_ARN设置RoleArn
           role_arn='<RoleArn>',
           # 角色会话名称，可以通过环境变量ALIBABA_CLOUD_ROLE_SESSION_NAME设置RoleSessionName
           role_session_name='<RoleSessionName>',
           # 设置更小的权限策略，非必填。示例值：{"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
           policy='<Policy>',
           # 设置角色会话有效期，单位为秒，默认值为3600秒（1小时），非必填
           role_session_expiration=3600
       )
   
       cred_client = Client(config)
   
       def get_credentials_wrapper():
           cred = cred_client.get_credential()
           return oss.credentials.Credentials(access_key_id=cred.access_key_id, access_key_secret=cred.access_key_secret, security_token=cred.security_token)
   
       # 创建凭证提供者，用于动态加载凭证
       credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
   
       # 加载OSS SDK的默认配置
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # 填写Bucket所在地域
       cfg.region = '<region-id>'
   
       # 创建OSS客户端实例
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   # 当此脚本被直接运行时，调用main函数
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

使用ECSRAMRole

如果您的应用程序运行在ECS实例、ECI实例、容器服务Kubernetes版的Worker节点中，建议您使用ECSRAMRole初始化凭证提供者。该方式底层实现是STS Token。ECSRAMRole允许您将一个角色关联到ECS实例、ECI实例或容器服务 Kubernetes 版的Worker节点，实现在实例内部自动刷新STS Token。该方式无需您提供一个AK或STS Token，消除了手动维护AK或STS Token的风险。如何获取ECSRAMRole，请参见创建角色。

1. 添加alibabacloud_credentials依赖。

   pip install alibabacloud_credentials

2. 配置ECSRAMRole作为访问凭证。

   运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

   from alibabacloud_credentials.client import Client
   from alibabacloud_credentials.models import Config
   import alibabacloud_oss_v2 as oss
   
   def main():
       config = Config(
           type='ecs_ram_role',      # 访问凭证类型。固定为ecs_ram_role。
           role_name='EcsRoleExample'    # 为ECS授予的RAM角色的名称。可选参数。如果不设置，将自动检索。强烈建议设置，以减少请求。
       )
   
       cred_client = Client(config)
   
       def get_credentials_wrapper():
           cred = cred_client.get_credential()
           return oss.credentials.Credentials(access_key_id=cred.access_key_id, access_key_secret=cred.access_key_secret, security_token=cred.security_token)
   
       # 创建凭证提供者，用于动态加载凭证
       credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
   
       # 加载OSS SDK的默认配置
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # 填写Bucket所在地域
       cfg.region = '<region-id>'
   
       # 创建OSS客户端实例
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   # 当此脚本被直接运行时，调用main函数
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

使用OIDCRoleARN

在容器服务Kubernetes版中设置了Worker节点RAM角色后，对应节点内的Pod中的应用也就可以像ECS上部署的应用一样，通过元数据服务（Meta Data Server）获取关联角色的STS Token。但如果容器集群上部署的是不可信的应用（比如部署您的客户提交的应用，代码也没有对您开放），您可能并不希望它们能通过元数据服务获取Worker节点关联实例RAM角色的STS Token。为了避免影响云上资源的安全，同时又能让这些不可信的应用安全地获取所需的STS Token，实现应用级别的权限最小化，您可以使用RRSA（RAM Roles for Service Account）功能。该方式底层实现是STS Token。阿里云容器集群会为不同的应用Pod创建和挂载相应的服务账户OIDC Token文件，并将相关配置信息注入到环境变量中，Credentials工具通过获取环境变量的配置信息，调用STS服务的AssumeRoleWithOIDC接口换取绑定角色的STS Token。该方式无需您提供一个AK或STS Token，消除了手动维护AK或STS Token的风险。详情请参见通过RRSA配置ServiceAccount的RAM权限实现Pod权限隔离。

使用自定义访问凭证

当以上凭证配置方式不满足要求时，您可以自定义获取凭证的方式。SDK 支持多种实现方式。

1. 通过credentials.CredentialsProviderFunc接口

   运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

   import argparse
   import alibabacloud_oss_v2 as oss
   import os
   
   def main():
   
       def get_credentials_wrapper():
           # 返回长期凭证
           return oss.credentials.Credentials(access_key_id='access_key_id', access_key_secret='access_key_security')
           # 返回STS临时凭证
           # return oss.credentials.Credentials(access_key_id='access_key_id', access_key_secret='access_key_security', security_token='security_token')
   
       credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
   
      # 加载SDK的默认配置，并设置凭证提供者
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
       
       # 填写Bucket所在地域
       cfg.region = "<region-id>"
   
       # 使用配置好的信息创建OSS客户端
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

2. 实现credentials.CredentialsProvider接口

   运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

   # -*- coding: utf-8 -*-
   import alibabacloud_oss_v2 as oss
   
   class CredentialProviderWrapper(oss.credentials.CredentialsProvider):
       def get_credentials(self):
           # TODO
           # 自定义访问凭证的获取方法
   
           # 返回长期凭证access_key_id, access_key_secrect
           return oss.credentials.Credentials('<access_key_id>', '<access_key_secrect>')
   
           # 返回 STS临时凭证access_key_id, access_key_secrect, token
           # 对于临时凭证，需要根据过期时间，刷新凭证。
           # return oss.credentials.Credentials('<access_key_id>', '<access_key_secrect>', '<token>');
   
   def main():
       # 创建凭证提供者，用于动态加载凭证
       credentials_provider = CredentialProviderWrapper()
   
       # 加载OSS SDK的默认配置
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # 填写Bucket所在地域
       cfg.region = '<region-id>'
   
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

匿名访问

如果您只需要访问公共读取权限的OSS资源，可以使用匿名访问方式，无需提供任何凭证。

运行示例代码前，请将代码中的 <region-id>占位符替换为实际的地域和Endpoint，如 cn-hangzhou。

import alibabacloud_oss_v2 as oss

# 创建匿名凭证提供者
credentials_provider = oss.credentials.AnonymousCredentialsProvider()

# 配置客户端
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# 填写Bucket所在地域
cfg.region = '<region-id>'

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

错误自助排查

使用OSS Python SDK V2访问OSS出现错误时，OSS会返回HTTP Code、Message、RequestId、EC错误码等信息，其中EC码对应一个具体的错误原因，您可以使用EC码自助进行错误排查。

示例代码

OSS Python SDK V2提供丰富的示例代码供参考或直接使用。