对象存储 OSS：OSS Python SDK V2

快速接入

接入OSS Python SDK V2的流程如下：

环境准备

要求 Python 3.8 及以上版本。

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

安装SDK

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

  pip install alibabacloud-oss-v2

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

  import alibabacloud_oss_v2 as oss

初始化客户端

使用地域和访问域名初始化 OSSClient，并运行测试代码。

import alibabacloud_oss_v2 as oss

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

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

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

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

    # 方式一：只填写Region（推荐）
    # 必须指定Region ID，以华东1（杭州）为例，Region填写为cn-hangzhou，SDK会根据Region自动构造HTTPS访问域名
    cfg.region = 'cn-hangzhou' 

    # # 方式二：同时填写Region和Endpoint
    # # 必须指定Region ID，以华东1（杭州）为例，Region填写为cn-hangzhou
    # cfg.region = 'cn-hangzhou'
    # # 填写Bucket所在地域对应的公网Endpoint。以华东1（杭州）为例，Endpoint填写为'https://oss-cn-hangzhou.aliyuncs.com'
    # cfg.endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'

    # 使用配置好的信息创建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函数

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

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

客户端配置

使用自定义域名

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

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 请填写您的自定义域名。例如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函数

超时控制

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

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

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

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

重试策略

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

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

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

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

HTTP/HTTPS 协议

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

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

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

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

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

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

代理服务器

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

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 设置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资源，可以降低流量成本并提高访问速度。

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'
    cfg.use_internal_endpoint = True

    # # 方式二： 直接填写Region和Endpoint
    # # 填写Bucket所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    # cfg.region = 'cn-hangzhou'
    # # 填写Bucket所在地域对应的内网Endpoint。以华东1（杭州）为例，Endpoint填写为'oss-cn-hangzhou-internal.aliyuncs.com'
    # cfg.endpoint = 'oss-cn-hangzhou-internal.aliyuncs.com'

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

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

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

使用传输加速域名

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'
    cfg.use_accelerate_endpoint = True

    # # 方式二： 直接填写Region和传输加速Endpoint
    # # 填写Bucket所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    # cfg.region = 'cn-hangzhou'
    # # 填写Bucket所在地域对应的传输加速Endpoint，例如'https://oss-accelerate.aliyuncs.com'
    # cfg.endpoint = 'https://oss-accelerate.aliyuncs.com'

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

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

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

使用专有域

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 请填写您的专有域。例如：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 客户端。

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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 使用配置好的信息创建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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 使用配置好的信息创建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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
    cfg.region = 'cn-hangzhou'

    # 使用配置好的信息创建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作为访问凭证。

   # -*- 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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
       cfg.region = 'cn-hangzhou'
   
       # 创建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，请参见CreateRole - 创建角色。

1. 添加alibabacloud_credentials依赖。

   pip install alibabacloud_credentials

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

   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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou。
       cfg.region = 'cn-hangzhou'
   
       # 创建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接口

   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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
       cfg.region = "cn-hangzhou"
   
       # 使用配置好的信息创建OSS客户端
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

2. 实现credentials.CredentialsProvider接口

   # -*- 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所在地域。以华东1（杭州）为例，Region填写为cn-hangzhou
       cfg.region = 'cn-hangzhou'
   
       client = oss.Client(cfg)
   
       # 使用client进行后续操作...
   
   
   if __name__ == "__main__":
       main()  # 脚本入口，当文件被直接运行时调用main函数

错误自助排查

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

示例代码

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