Object Storage Service:OSS Python SDK V2

Configurar credenciais de acesso

Configure as credenciais com um par de AccessKey de um usuário RAM.

1. No console do RAM, crie um usuário RAM e selecione Using permanent AccessKey to access. Salve o par de AccessKey e conceda a permissão AliyunOSSFullAccess ao usuário.

2. Use o par de AccessKey do usuário RAM para configurar as variáveis de ambiente.

   Linux

   1. Anexe as variáveis de ambiente ao arquivo ~/.bashrc.

      echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bashrc
      echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bashrc

      1. Aplique as alterações.

         source ~/.bashrc

      2. Verifique as variáveis de ambiente.

         echo $OSS_ACCESS_KEY_ID
         echo $OSS_ACCESS_KEY_SECRET

   macOS

   1. Verifique o tipo de shell padrão.

      echo $SHELL

      1. Siga as etapas correspondentes ao seu tipo de shell.

         Zsh

         1. Anexe as variáveis de ambiente ao arquivo ~/.zshrc.

            echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.zshrc
            echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.zshrc

         2. Aplique as alterações.

            source ~/.zshrc

         3. Verifique as variáveis de ambiente.

            echo $OSS_ACCESS_KEY_ID
            echo $OSS_ACCESS_KEY_SECRET

         Bash

         1. Anexe as variáveis de ambiente ao arquivo ~/.bash_profile.

            echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bash_profile
            echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bash_profile

         2. Aplique as alterações.

            source ~/.bash_profile

         3. Verifique as variáveis de ambiente.

            echo $OSS_ACCESS_KEY_ID
            echo $OSS_ACCESS_KEY_SECRET

   Windows

   CMD

   1. Defina as variáveis de ambiente no CMD.

      setx OSS_ACCESS_KEY_ID "YOUR_ACCESS_KEY_ID"
      setx OSS_ACCESS_KEY_SECRET "YOUR_ACCESS_KEY_SECRET"

      1. Verifique as variáveis de ambiente.

         echo %OSS_ACCESS_KEY_ID%
         echo %OSS_ACCESS_KEY_SECRET%

   PowerShell

   1. Defina as variáveis de ambiente no PowerShell.

      [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_ID", "YOUR_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
      [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", "YOUR_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)

      1. Verifique as variáveis de ambiente.

         [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
         [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)

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

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

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

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

#!/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

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

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

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

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

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

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

1. Adicione a dependência alibabacloud_credentials.

   pip install alibabacloud_credentials

1. Configure o ARN da função OIDC como credencial de acesso.

   Antes de executar o código de exemplo, substitua o espaço reservado <region-id> por uma região e endpoint , por exemplo, ou 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(
           # Specify the credential type. The value is fixed to 'oidc_role_arn'.
           type='oidc_role_arn',
           # The ARN of the RAM role. You can set this value by using the ALIBABA_CLOUD_ROLE_ARN environment variable.
           role_arn=os.environ.get('<RoleArn>'),
           # The OIDC provider ARN. You can set this value by using the ALIBABA_CLOUD_OIDC_PROVIDER_ARN environment variable.
           oidc_provider_arn=os.environ.get('<OidcProviderArn>'),
           # The OIDC token file path. You can set this value by using the ALIBABA_CLOUD_OIDC_TOKEN_FILE environment variable.
           oidc_token_file_path=os.environ.get('<OidcTokenFilePath>'),
           # The role session name. You can set this value by using the ALIBABA_CLOUD_ROLE_SESSION_NAME environment variable.
           role_session_name='<RoleSessionName>',
           # A more restrictive permission policy. This parameter is optional. Example: {"Statement": [{"Action": ["*"],"Effect": "Allow","Resource": ["*"]}],"Version":"1"}
           policy='<Policy>',
           # The validity period of the role session in seconds. The default is 3600 seconds (1 hour). This parameter is optional.
           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 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...
   
   # Call the main function when this script is run directly.
   if __name__ == "__main__":
       main()  # Script entry point. Calls the main function when the file is run directly.

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)

1. Por exemplo, o código a seguir dispara um erro ao tentar baixar um objeto inexistente.

   Antes de executar o código de exemplo, substitua espaços reservados como <region-id> por uma região e endpoint reais, por exemplo, ou 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
   
       # Set the region in the configuration.
       cfg.region = '<region-id>'
   
       # Create an OSS client with the specified configurations.
       client = oss.Client(cfg)
   
       # Send a request to get the object, specifying the bucket name and object key.
       result = client.get_object(oss.GetObjectRequest(
           bucket="Your Bucket Name",  # Specify the bucket name.
           key="Your Object Key",  # Specify the object key.
       ))
   
       # Print the status code and request ID from the response.
       print(f'status code: {result.status_code},'
             f' request id: {result.request_id},'
       )
   
       # Use a context manager to ensure the resource is released.
       with result.body as body_stream:
           data = body_stream.read()
           print(f"Object read complete. Data length: {len(data)} bytes")
   
           path = "./get-object-sample.txt"
           with open(path, 'wb') as f:
               f.write(data)
           print(f"Object downloaded and saved to: {path}")
   
   if __name__ == "__main__":
       main()  # Script entry point.

2. A execução do script retorna o seguinte erro. A resposta inclui a linha 'EC': '0026-00000001', que fornece um identificador exclusivo para a causa do erro.

   alibabacloud_oss_v2.exceptions.OperationError: operation error GetObject: Error returned by Service.
           Http Status Code: 404.
           Error Code: NoSuchKey.
           Request Id: 67CA9AD3ECB4DB3034E5A92D.
           Message: The specified key does not exist..
           EC: 0026-00000001.
           Timestamp: 2025-03-07 07:05:55+00:00.
           Request Endpoint: GET https://xxx.oss-cn-hangzhou.aliyuncs.com/asghdjfjdas.

3. Para diagnosticar o erro, use o EC da saída de erro e siga estas etapas.

   1. Abra a Plataforma de Diagnóstico Self-service OpenAPI.

   2. Na caixa de pesquisa, insira o EC, por exemplo, 0026-00000001, e clique em Diagnose.

   3. Revise os resultados da pesquisa para encontrar a causa e a solução do erro. A página de diagnóstico detalha a Problem Description (o objeto alvo não existe) e lista Possible Causes (por exemplo, falha no upload do objeto ou exclusão por um usuário, regra de ciclo de vida ou replicação entre regiões), um Problem Example e uma Solution. A solução pode sugerir o uso da interface HeadObject para verificar se o objeto existe, o uso da interface ListObjectVersions para verificar o VersionID, ou a revisão de configurações de convenções de nomenclatura, regras de ciclo de vida, permissões de usuário e replicação entre regiões. A seção References na parte inferior contém links para a documentação sobre NoSuchKey e NoSuchVersion.