Object Storage Service:OSS Python SDK V2

Quick start

Prerequisites

You need Python 3.8 or later.

To check your Python version, run the python --version command. If Python is not installed or your version is lower than 3.8, download and install Python.

Install the SDK

• Install the latest version to ensure the code examples run correctly.

  pip install alibabacloud-oss-v2

• Import the SDK package:

  import alibabacloud_oss_v2 as oss

Initialize client

Before you run the sample code, replace placeholders such as <region-id> with the actual region and endpoint, such as ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    """
    Configuration details for initializing the Python SDK V2 client:

    1. Signature version: The SDK uses Signature V4 by default for enhanced security.
    2. Region configuration: When you initialize the client, you must specify an Alibaba Cloud region ID.
    3. Endpoint configuration:
       - You can use the endpoint parameter to specify a custom endpoint for service requests.
       - If you do not specify an endpoint, the SDK automatically constructs a public endpoint based on the specified region.
    4. Protocol configuration:
       - The SDK uses HTTPS to construct the endpoint by default.
       - To use HTTP, you must explicitly specify the protocol in the endpoint.
    """

    # Load credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Method 1: Specify only the region (Recommended).
    # You must specify a region ID. The SDK automatically constructs an HTTPS endpoint based on the region.
    cfg.region = '<region-id>' 

    # # Method 2: Specify both the region and the endpoint.
    # # You must specify a region ID.
    # cfg.region = '<region-id>'
    # # Specify the public endpoint of the region where the bucket is located.
    # cfg.endpoint = '<endpoint>'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Define the string to upload.
    text_string = "Hello, OSS!"
    data = text_string.encode('utf-8')  # Encode the string as a UTF-8 byte string.

    # Send a request to upload the object. Specify the bucket name, object key, and data.
    result = client.put_object(oss.PutObjectRequest(
        bucket="Your Bucket Name",
        key="Your Object Key",
        body=data,
    ))

    # Print the status code, request ID, and ETag to verify that the request was successful.
    print(f'status code: {result.status_code}\n'
          f'request id: {result.request_id}\n'
          f'etag: {result.etag}'
    )

# When this script is run directly, call the main function.
if __name__ == "__main__":
    main()

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

async def main():
    """
    Configuration details for initializing the Python SDK V2 async client:

    1. Signature version: The SDK uses Signature V4 by default for enhanced security.
    2. Region configuration: When you initialize the AsyncClient, you must specify an Alibaba Cloud region ID.
    3. Endpoint configuration:
       - You can use the endpoint parameter to specify a custom endpoint for service requests.
       - If you do not specify an endpoint, the SDK automatically constructs a public endpoint based on the specified region.
    4. Protocol configuration:
       - The SDK uses HTTPS to construct the endpoint by default.
       - To use HTTP, you must explicitly specify the protocol in the endpoint.
    5. Asynchronous features:
       - Import the async module: import alibabacloud_oss_v2.aio as oss_aio
       - Create an async client: oss_aio.AsyncClient(cfg)
       - All operations require the await keyword.
       - You must call await client.close() in a finally block to close the connection.
       - You must install the aiohttp dependency: pip install aiohttp
    """

    # Load credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Method 1: Specify only the region (Recommended).
    # You must specify a region ID. The SDK automatically constructs an HTTPS endpoint based on the region.
    cfg.region = '<region-id>'

    # # Method 2: Specify both the region and the endpoint.
    # # You must specify a region ID.
    # cfg.region = '<region-id>'
    # # Specify the public endpoint of the region where the bucket is located.
    # cfg.endpoint = '<endpoint>'

    # Create an OSS async client with the specified configurations.
    client = oss_aio.AsyncClient(cfg)

    try:
        # Define the string to upload.
        text_string = "Hello, OSS!"
        data = text_string.encode('utf-8')  # Encode the string as a UTF-8 byte string.

        # Send an async request to upload the object. Specify the bucket name, object key, and data.
        # Note: Use the await keyword to wait for the async operation to complete.
        result = await client.put_object(
            oss.PutObjectRequest(
                bucket="Your Bucket Name",
                key="Your Object Key",
                body=data,
            )
        )

        # Print the status code, request ID, and ETag to verify that the request was successful.
        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'Upload failed: {e}')

    finally:
        # Close the async client connection to prevent resource leaks.
        await client.close()

# When this script is run directly, call the main function.
if __name__ == "__main__":
    # Use asyncio.run() to run the async main function.
    asyncio.run(main())

Successful upload output:

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

Client configuration

Custom domain name

Custom domain names enable object previews in a browser and CDN-accelerated delivery.

Before running the sample code, replace the <region-id> placeholder with the region and endpoint, for example, ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the bucket's region.
    cfg.region = '<region-id>'

    # Specify your custom domain name. For example, www.example-***.com.
    cfg.endpoint = 'https://www.example-***.com'

    # Set use_cname to True to use a custom domain name.
    cfg.use_cname = True

    # Create an OSS client.
    client = oss.Client(cfg)

    # Use the client for subsequent operations.

# Script entry point.
if __name__ == "__main__":
    main()

Timeout control

Before you run the sample code, replace the <region-id> placeholder with a region and endpoint, such as ap-southeast-1.

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

def main():
    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configuration.
    cfg = oss.config.load_default()

    # Override the default connection timeout (10 seconds).
    cfg.connect_timeout = 30

    # Override the default read/write timeout (20 seconds).
    cfg.readwrite_timeout = 30

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Specify the region.
    cfg.region = '<region-id>'

    # Create an OSS client with the specified configuration.
    client = oss.Client(cfg)

# Run the main function when the script is executed directly.
if __name__ == "__main__":
    main()

Retry policy

Before running the sample code, replace the <region-id> placeholder with the region and endpoint, for example, ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    """
    SDK retry policy configuration:

    Default retry policy:
    If you do not configure a retry policy, the SDK uses StandardRetryer() by default. The default configuration is as follows:
    - max_attempts: The maximum number of retry attempts. Default: 3.
    - max_backoff: The maximum backoff time in seconds. Default: 20.
    - base_delay: The base delay time in seconds. Default: 0.2.
    - backoff_delayer: The backoff algorithm. The SDK uses exponential backoff with jitter (FullJitter) by default.
      Formula: random.uniform(0, 1) * min(2 ** attempts * base_delay, max_backoff)
    - error_retryables: The types of retryable errors. For more information, see https://gosspublic.alicdn.com/sdk-doc/alibabacloud-oss-python-sdk-v2/latest/_modules/alibabacloud_oss_v2/retry/error_retryable.html

    When a retryable error occurs, the client delays and retries the request according to this configuration.
    The overall request latency increases with each retry. If the default configuration does not meet your requirements,
    you can customize the retry parameters or modify the retry implementation.
    """

    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configurations.
    cfg = oss.config.load_default()

    # Retry policy configuration examples:

    # 1. Customize the maximum number of retries (default: 3, set to 5 here).
    cfg.retryer = oss.retry.StandardRetryer(max_attempts=5)

    # 2. Customize the backoff delay time.
    # Adjust base_delay to 0.5 seconds (default: 0.2) and max_backoff to 25 seconds (default: 20).
    # cfg.retryer = oss.retry.StandardRetryer(max_backoff=25, base_delay=0.5)

    # 3. Customize the backoff algorithm.
    # Replace the default FullJitter algorithm with a fixed-delay backoff of 2 seconds.
    # cfg.retryer = oss.retry.StandardRetryer(backoff_delayer=oss.retry.FixedDelayBackoff(2))

    # 4. Disable the retry policy.
    # Use retry.NopRetryer() to disable all retry attempts.
    # cfg.retryer = oss.retry.NopRetryer()

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Use the created client for subsequent operations...

# When this script is run directly, call the main function.
if __name__ == "__main__":
    main()

Set the connection pool size

For the http_client, use the max_connections parameter to set the connection pool size.

Before you run the sample code, replace the <region-id> placeholder with an actual region, for example, ap-southeast-1.

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

# OSS Python SDK V2 connection pool configuration example

import alibabacloud_oss_v2 as oss

def main():

    # Load access credentials from environment variables (OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET must be set).
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configuration and set the credentials provider.
    config = oss.config.load_default()
    config.credentials_provider = credentials_provider

    # Set the OSS service region.
    config.region = "<region-id>"

    # ============ Custom HTTP client configuration ============
    # Create a dictionary for the HTTP client configuration.
    http_client_config = {}
    
    # Set the connection pool size to 100. The default is 20.
    http_client_config["max_connections"] = 100
    
    # Create an HTTP client with the custom configuration.
    http_client = oss.transport.RequestsHttpClient(**http_client_config)
    
    # Set the custom HTTP client in the configuration object.
    config.http_client = http_client
    # ============================================
    
    # Initialize the OSS client instance.
    client = oss.Client(config)

    # Configure bucket and object information.
    bucket = "example-bucket"             # bucket name
    key = "dest.jpg"                      # object key in OSS
    file_path = "dest.jpg"                # Local save path

    # Download the object from OSS to the specified local path.
    client.get_object_to_file(oss.GetObjectRequest(bucket, key), file_path)
    print(f"Object download complete: {key} -> {file_path}")
    print(f"Connection pool size set to: {http_client_config['max_connections']}")

if __name__ == "__main__":
    # Program entry point
    try:
        main()
    except Exception as e:
        print(f"An error occurred during execution: {e}")
        raise

HTTP/HTTPS

Use cfg.disable_ssl = True to disable HTTPS requests.

Before running the sample code, replace the <region-id> placeholder with a region and endpoint, such as or ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configuration and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the client region.
    cfg.region = '<region-id>'

    # Disable HTTPS requests.
    cfg.disable_ssl = True

    # Create an OSS client using the specified configuration.
    client = oss.Client(cfg)

    # Use the created client to perform subsequent operations...

# The following entry point calls the main function when the script is run directly.
if __name__ == "__main__":
    main()

Proxy server

Configure a proxy server when corporate policies restrict direct internet access.

Before running the sample code, replace the <region-id> placeholder with a valid region and endpoint, for example, or ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'

    # Set the user agent.
    cfg.user_agent = 'aliyun-sdk-python'

    # Set the proxy server.
    cfg.proxy_host = 'http://user:passswd@proxy.example-***.com'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Use the created client to perform subsequent operations...

# When this script is run directly, call the main function.
if __name__ == "__main__":
    main()

Use an internal endpoint

Access OSS resources in the same region through an internal endpoint to reduce costs and latency.

Before you run the sample code, replace placeholders such as <region-id> with your actual region and endpoint values, such as ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configuration and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Method 1: Specify the region and set use_internal_endpoint to true.
    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'
    cfg.use_internal_endpoint = True

    # # Method 2: Specify the region and endpoint directly.
    # # Specify the region where the bucket is located.
    # cfg.region = '<region-id>'
    # # Specify the internal endpoint for the bucket's region.
    # cfg.endpoint = '<endpoint>'

    # Create the OSS client.
    client = oss.Client(cfg)

    # Use the client for subsequent operations...

# Call the main function if the script is executed directly.
if __name__ == "__main__":
    main()

Transfer acceleration

Before you run the sample code, replace the <region-id> placeholder with a region and endpoint, for example, or ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Method 1: Specify the region and set use_accelerate_endpoint to true.
    # Specify the bucket's region.
    cfg.region = '<region-id>'
    cfg.use_accelerate_endpoint = True

    # # Method 2: Directly specify the region and the acceleration endpoint.
    # # Specify the bucket's region.
    # cfg.region = '<region-id>'
    # # Specify the acceleration endpoint.
    # cfg.endpoint = 'https://oss-accelerate.aliyuncs.com'

    # Create an OSS client with the configuration.
    client = oss.Client(cfg)

    # Use this client for subsequent operations...

# Run the main function if the script is executed directly.
if __name__ == "__main__":
    main()

Use a dedicated domain

Before running the sample code, replace the <region-id> placeholder with your region and endpoint, such as ap-southeast-1.

import alibabacloud_oss_v2 as oss

def main():
    # Load credentials from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default SDK configurations and set the credentials provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'

    # Specify your dedicated domain. For example, https://service.corp.example.com.
    cfg.endpoint = 'https://service.corp.example.com'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Use the created client to perform subsequent operations...

# When this script is run directly, call the main function.
if __name__ == "__main__":
    main()

Custom HTTP client

Replace the default HTTP client with cfg.http_client when standard configuration parameters are insufficient.

Before running the sample code, replace the <region-id> placeholder with a valid region and endpoint, for example, or ap-southeast-1.

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

def main():
    # Load credentials from environment variables for authentication.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # Load the default configuration.
    cfg = oss.config.load_default()

    # Set the parameters for the HTTP client.
    kwargs: Dict[str, Any] = {}

    # Set the session.
    # kwargs["session"] = requests.Session()

    # Set the adapter.
    # kwargs["adapter"] = HTTPAdapter()

    # Specifies whether to skip certificate verification. Default: false.
    # kwargs["insecure_skip_verify"] = False

    # Specifies whether to enable HTTP redirection. Default: false.
    # kwargs["enabled_redirect"] = False

    # Set the proxy server.
    # kwargs["proxy_host"] = config.proxy_host

    # Set the block size.
    # kwargs["block_size"] = 16 * 1024

    # Connection timeout in seconds. Default: 10.
    kwargs["connect_timeout"] = 30

    # Read-write timeout in seconds. Default: 20.
    kwargs["readwrite_timeout"] = 30

    # Maximum connections. Default: 20.
    kwargs["max_connections"] = 1024

    # Create an HTTP client with the specified parameters.
    cfg.http_client = oss.transport.RequestsHttpClient(**kwargs)

    # Set the credentials provider.
    cfg.credentials_provider = credentials_provider

    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'

    # Create an OSS client with the specified configuration.
    client = oss.Client(cfg)

    # Use the client for subsequent OSS operations...

# Calls the main function when the script is executed directly.
if __name__ == "__main__":
    main()

Access credential configuration

Choose a credential method for your use case.

Use a RAM user's AccessKey

Initialize the credential provider with an AccessKey pair (AccessKey ID and AccessKey Secret) of your Alibaba Cloud account or a RAM user. This method requires manual key management and carries higher security risk.

import alibabacloud_oss_v2 as oss

def main():
    # Create a static credential provider and explicitly set the AccessKey ID and AccessKey secret.
    # Replace the placeholders with the AccessKey ID and AccessKey secret of your RAM user.
    credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id="RAM AccessKey ID",
        access_key_secret="RAM AccessKey Secret"
    )

    # Load the default SDK configurations and set the credential provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the region where the bucket is located.
    cfg.region = '<region-id>'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Use the created client to perform subsequent operations...

# When this script is run directly, call the main function.
if __name__ == "__main__":
    main()  # Script entry point. Calls the main function when the file is run directly.

Use temporary credentials from STS

Initialize the credential provider with temporary STS credentials (AccessKey ID, AccessKey Secret, and Security Token). Tokens must be refreshed manually before expiration.

import alibabacloud_oss_v2 as oss

def main():
    # Use the temporary AccessKey ID and AccessKey Secret from STS, not the credentials of your Alibaba Cloud account.
    # Note that the AccessKey ID from STS starts with "STS.".
    sts_access_key_id = 'STS.****************'
    sts_access_key_secret = 'yourAccessKeySecret'
    # Provide the Security Token obtained from STS.
    sts_security_token = 'yourSecurityToken'

    # Create a static credential provider and explicitly set the temporary AccessKey ID, AccessKey Secret, and Security Token.
    credentials_provider = oss.credentials.StaticCredentialsProvider(
        access_key_id=sts_access_key_id,
        access_key_secret=sts_access_key_secret,
        security_token=sts_security_token,
    )

    # Load the default SDK configurations and set the credential provider.
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Specify the region where the Bucket is located.
    cfg.region = '<region-id>'

    # Create an OSS client with the specified configurations.
    client = oss.Client(cfg)

    # Use the created client to perform subsequent operations...

# Script entry point
if __name__ == "__main__":
    main()  # Calls the main function.

Use a RAM role ARN

Initialize the credential provider with a RAM role ARN for delegated access (such as cross-account). The SDK automatically refreshes the STS token via AssumeRole. Use the policy parameter to restrict permissions.

1. Add the alibabacloud_credentials dependency.

   pip install alibabacloud_credentials

2. Configure the AccessKey pair and RAM role ARN as access credentials.

   Before you run the sample code, replace the <region-id> placeholder with the actual region and endpoint, for example, or ap-southeast-1.

   # -*- 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(
           # The RAM user's AccessKey ID and AccessKey secret, retrieved from environment variables.
           access_key_id=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
           access_key_secret=os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
           type='ram_role_arn',
           # The ARN of the RAM role to assume. Example: acs:ram::123456789012****:role/adminrole. You can set this using the ALIBABA_CLOUD_ROLE_ARN environment variable.
           role_arn='<RoleArn>',
           # The role session name. You can set this using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
           role_session_name='<RoleSessionName>',
           # Optional. A more restrictive permission policy. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
           policy='<Policy>',
           # Optional. The session duration in seconds. Default: 3600 (1 hour).
           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)
   
       # Create a credential provider to dynamically load credentials.
       credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
   
       # Load the default OSS SDK configurations.
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # Specify the region where the bucket is located.
       cfg.region = '<region-id>'
   
       # Create an OSS client instance.
       client = oss.Client(cfg)
   
       # Use the client for subsequent operations...
   
   if __name__ == "__main__":
       main()  # Script entry point. Calls the main function when the file is run directly.

Use an ECS instance RAM role

For applications on ECS instances, ECI instances, or Container Service for Kubernetes worker nodes, use an ECS instance RAM role. The SDK automatically refreshes the STS token without hardcoded credentials. Create a RAM role.

1. Add the alibabacloud_credentials dependency.

   pip install alibabacloud_credentials

2. Configure the ECS instance RAM role as the access credential.

   Before you run the sample code, replace the <region-id> placeholder with an actual region ID, for example, or ap-southeast-1.

   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',      # Specifies the access credential type. Must be 'ecs_ram_role'.
           role_name='EcsRoleExample'    # Optional: The name of the RAM role granted to the ECS instance. If omitted, the role name is automatically retrieved. We recommend that you set this parameter to reduce requests.
       )
   
       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)
   
       # Create a credentials provider to dynamically load credentials.
       credentials_provider = oss.credentials.CredentialsProviderFunc(func=get_credentials_wrapper)
   
       # Load the default OSS SDK configurations.
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # Specify the region where the bucket is located.
       cfg.region = '<region-id>'
   
       # Create an OSS client instance.
       client = oss.Client(cfg)
   
       # Use the client for subsequent operations...
   
   if __name__ == "__main__":
       main()

OIDC role ARN

For untrusted applications on Container Service for Kubernetes worker nodes, use RRSA (RAM Roles for Service Accounts) to enforce least-privilege at the pod level. RRSA mounts an OIDC token file per pod and calls AssumeRoleWithOIDC to exchange it for a scoped STS token. Isolate pod permissions based on RRSA.

Custom access credentials

Customize credential retrieval with one of these approaches:

1. Using the credentials.CredentialsProviderFunc interface

   Before you run the sample code, replace the <region-id> placeholder with the actual region and endpoint, for example, or ap-southeast-1.

   import argparse
   import alibabacloud_oss_v2 as oss
   import os
   
   def main():
   
       def get_credentials_wrapper():
           # Return long-term credentials.
           return oss.credentials.Credentials(access_key_id='access_key_id', access_key_secret='access_key_security')
           # Return temporary STS credentials.
           # 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)
   
      # Load the default SDK configurations and set the credentials provider.
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
       
       # Specify the region where the bucket is located.
       cfg.region = "<region-id>"
   
       # Create an OSS client with the specified configurations.
       client = oss.Client(cfg)
   
       # Use the client to perform subsequent operations...
   
   if __name__ == "__main__":
       main()  # Script entry point. Calls the main function when the file is run directly.

2. Implementing the credentials.CredentialsProvider interface

   Before you run the sample code, replace the <region-id> placeholder with the actual region and endpoint, for example, or ap-southeast-1.

   # -*- coding: utf-8 -*-
   import alibabacloud_oss_v2 as oss
   
   class CredentialProviderWrapper(oss.credentials.CredentialsProvider):
       def get_credentials(self):
           # TODO
           # Customize the logic for obtaining access credentials.
   
           # Return long-term credentials: access_key_id, access_key_secret.
           return oss.credentials.Credentials('<access_key_id>', '<access_key_secret>')
   
           # Return temporary STS credentials: access_key_id, access_key_secret, token.
           # Temporary credentials must be refreshed before they expire.
           # return oss.credentials.Credentials('<access_key_id>', '<access_key_secret>', '<token>');
   
   def main():
       # Instantiate a custom credentials provider.
       credentials_provider = CredentialProviderWrapper()
   
       # Load the default OSS SDK configurations.
       cfg = oss.config.load_default()
       cfg.credentials_provider = credentials_provider
   
       # Specify the region where the bucket is located.
       cfg.region = '<region-id>'
   
       client = oss.Client(cfg)
   
       # Use the client to perform subsequent operations...
   
   if __name__ == "__main__":
       main()  # Script entry point. Calls the main function when the file is run directly.

Anonymous access

Access public-read OSS resources without credentials.

Before you run the sample code, replace the <region-id> placeholder with an actual region and endpoint, for example, or ap-southeast-1.

import alibabacloud_oss_v2 as oss

# Create an anonymous credentials provider.
credentials_provider = oss.credentials.AnonymousCredentialsProvider()

# Configure the client.
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# Specify the bucket's region.
cfg.region = '<region-id>'

# Create an OSS client.
client = oss.Client(cfg)

Troubleshoot errors

When a request fails, the response includes an HTTP status code, message, request ID, and error code (EC). Use the EC to diagnose the root cause.

Code examples

Common operation code examples: