Object Storage Service:OSS Python SDK V2

クイック統合

このセクションでは、OSS Python SDK V2 の統合方法について説明します。

前提条件

Python 3.8 以降が必要です。

python --version コマンドを実行して、Python のバージョンを確認できます。 Python がインストールされていない、またはバージョンが Python 3.8 より前の場合は、Python をダウンロードしてインストールしてください。

SDK のインストール

• 次のコマンドを実行して、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. リージョン設定：クライアントを初期化する際に、Alibaba Cloud のリージョン ID を指定する必要があります。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    3. エンドポイント設定：
       - endpoint パラメーターを使用して、サービスリクエストのエンドポイントをカスタマイズできます。
       - エンドポイントを指定しない場合、リージョンに基づいてパブリックエンドポイントが自動的に構築されます。たとえば、リージョンが cn-hangzhou の場合、エンドポイントは https://oss-cn-hangzhou.aliyuncs.com です。
    4. プロトコル設定：
       - SDK は、デフォルトで HTTPS を使用してエンドポイントを構築します。
       - HTTP を使用するには、エンドポイントでプロトコルを明示的に指定します。例：http://oss-cn-hangzhou.aliyuncs.com。
    """

    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 方法 1：リージョンのみを指定 (推奨)。
    # リージョン ID を指定する必要があります。たとえば、リージョンを中国 (杭州) の cn-hangzhou に設定すると、SDK は自動的に HTTPS エンドポイントを構築します。
    cfg.region = 'cn-hangzhou' 

    # # 方法 2：リージョンとエンドポイントの両方を指定。
    # # リージョン ID を指定する必要があります。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    # cfg.region = 'cn-hangzhou'
    # # バケットが配置されているリージョンのパブリックエンドポイントを指定します。たとえば、中国 (杭州) のエンドポイントは '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="ご利用のバケット名",
        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. リージョン設定：AsyncClient を初期化する際に、Alibaba Cloud のリージョン ID を指定する必要があります。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。

    3. エンドポイント設定：
       - endpoint パラメーターを使用して、サービスリクエストのエンドポイントをカスタマイズできます。
       - エンドポイントを指定しない場合、リージョンに基づいてパブリックエンドポイントが自動的に構築されます。たとえば、リージョンが cn-hangzhou の場合、エンドポイントは https://oss-cn-hangzhou.aliyuncs.com です。

    4. プロトコル設定：
       - SDK は、デフォルトで HTTPS を使用してエンドポイントを構築します。
       - HTTP を使用するには、エンドポイントでプロトコルを明示的に指定します。例：http://oss-cn-hangzhou.aliyuncs.com。
    
    5. 非同期機能：
       - 非同期モジュールをインポートします：import alibabacloud_oss_v2.aio as oss_aio
       - 非同期クライアントを作成します：oss_aio.AsyncClient(cfg)
       - すべての操作には await キーワードが必要です。
       - 接続を閉じるには、finally ブロックで await client.close() を呼び出す必要があります。
       - aiohttp 依存関係をインストールする必要があります：pip install aiohttp
    """

    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 方法 1：リージョンのみを指定 (推奨)。
    # リージョン ID を指定する必要があります。たとえば、リージョンを中国 (杭州) の cn-hangzhou に設定すると、SDK は自動的に HTTPS エンドポイントを構築します。
    cfg.region = 'cn-hangzhou'

    # # 方法 2：リージョンとエンドポイントの両方を指定。
    # # リージョン ID を指定する必要があります。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    # cfg.region = 'cn-hangzhou'
    # # バケットが配置されているリージョンのパブリックエンドポイントを指定します。たとえば、中国 (杭州) のエンドポイントは 'https://oss-cn-hangzhou.aliyuncs.com' です。
    # cfg.endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'

    # 指定された設定で 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="ご利用のバケット名",
                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'Upload failed: {e}')

    finally:
        # 非同期クライアント接続を閉じます (重要：これによりリソースリークが防止されます)。
        await client.close()


# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    # asyncio.run() を使用して非同期の main 関数を実行します。
    asyncio.run(main())

コードを実行すると、次の結果が返されます。これは、ファイルが正常にアップロードされたことを示します。

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

クライアント設定

カスタムドメイン名の使用

デフォルトの OSS エンドポイントを使用すると、オブジェクトにアクセスしたりプレビューしたりできない場合があります。カスタムドメイン名経由で OSS にアクセスする場合、ブラウザで直接オブジェクトをプレビューし、Alibaba Cloud CDN を使用してコンテンツ配信を高速化できます。

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # ご利用のカスタムドメイン名を指定します。例：www.example-***.com。
    cfg.endpoint = 'https://www.example-***.com'

    # CNAME オプションを有効にするには、このパラメーターを true に設定する必要があることに注意してください。そうしないと、カスタムドメイン名を使用できません。
    cfg.use_cname = True

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

タイムアウト制御

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    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

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、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 をご参照ください。

    リトライ可能なエラーが発生した場合、提供された設定を使用して遅延させ、リクエストをリトライします。
    全体的なリクエストのレイテンシはリトライ回数とともに増加します。デフォルトの設定がニーズに合わない場合は、
    リトライパラメーターを設定するか、リトライ実装を変更できます。
    """

    # ID 検証のために環境変数から認証情報を読み込みます。
    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

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

接続プールサイズの設定

http_client では、max_connections パラメーターを指定して接続プールサイズを設定できます。

#!/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 = "cn-hangzhou"

    # ============ カスタム 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 = "example-bucket"             # バケット名
    key = "dest.jpg"                      # OSS 内のオブジェクトのパス
    file_path = "dest.jpg"                # オブジェクトを保存するローカルパス

    # OSS から指定されたローカルパスにオブジェクトをダウンロードします。
    client.get_object_to_file(oss.GetObjectRequest(bucket, key), file_path)
    print(f"Download complete: {key} -> {file_path}")
    print(f"Connection pool size set to: {http_client_config['max_connections']}")

if __name__ == "__main__":
    # プログラムのエントリポイント。
    try:
        main()
    except Exception as e:
        print(f"An error occurred during execution: {e}")
        raise

HTTP/HTTPS プロトコル

cfg.disable_ssl = True を設定すると、HTTPS プロトコルを無効にできます。

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # HTTPS リクエストを使用しません。
    cfg.disable_ssl = True

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

プロキシサーバー

企業のセキュリティポリシーがパブリックネットワークへの直接アクセスを制限している場合、プロキシサーバーを設定して OSS にアクセスできます。

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は 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)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

同一リージョン内の内部エンドポイントの使用

同一リージョン内の内部エンドポイントを使用して、同じリージョン内の OSS リソースにアクセスすることで、トラフィックコストを削減し、アクセス速度を向上させることができます。

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 方法 1：リージョンを指定し、use_internal_endpoint を true に設定します。
    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'
    cfg.use_internal_endpoint = True

    # # 方法 2：リージョンとエンドポイントを直接指定します。
    # # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    # cfg.region = 'cn-hangzhou'
    # # バケットが配置されているリージョンの内部エンドポイントを指定します。たとえば、中国 (杭州) のエンドポイントは 'oss-cn-hangzhou-internal.aliyuncs.com' です。
    # cfg.endpoint = 'oss-cn-hangzhou-internal.aliyuncs.com'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

アクセラレーションエンドポイントの使用

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 方法 1：リージョンを指定し、use_accelerate_endpoint を true に設定します。
    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'
    cfg.use_accelerate_endpoint = True

    # # 方法 2：リージョンとアクセラレーションエンドポイントを直接指定します。
    # # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    # cfg.region = 'cn-hangzhou'
    # # バケットが配置されているリージョンのアクセラレーションエンドポイントを指定します。例：'https://oss-accelerate.aliyuncs.com'。
    # cfg.endpoint = 'https://oss-accelerate.aliyuncs.com'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

専用ドメインの使用

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # ご利用の専用ドメインを指定します。例：https://service.corp.example.com。
    cfg.endpoint = 'https://service.corp.example.com'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

ガバメントクラウドエンドポイントの使用

次のコードは、ガバメントクラウドエンドポイントを使用して OSSClient を設定する方法を示しています。

import alibabacloud_oss_v2 as oss

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの SDK 設定を読み込み、認証情報プロバイダーを設定します。
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # リージョンとエンドポイントを指定します。
    # バケットが配置されているリージョンを指定します。たとえば、中国 (北京) ガバメントクラウド 1 の場合、リージョンを cn-north-2-gov-1 に設定します。
    cfg.region = 'cn-north-2-gov-1'
    # バケットが配置されているリージョンの内部エンドポイントを指定します。たとえば、中国 (北京) ガバメントクラウド 1 の場合、エンドポイントは '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)

    # 作成されたクライアントを使用して後続の操作を実行します...

# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

HTTP クライアントのカスタマイズ

共通の設定パラメーターがニーズを満たさない場合は、cfg.http_client を使用してデフォルトの HTTP クライアントを置き換えることができます。

import alibabacloud_oss_v2 as oss
from typing import Dict, Any

def main():
    # ID 検証のために環境変数から認証情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # デフォルトの設定を読み込みます。
    cfg = oss.config.load_default()

    # HTTP クライアントのパラメーターを設定します。
    kwargs: Dict[str, Any] = {}

    # セッションを設定します。
    # kwargs["session"] = requests.Session()

    # アダプターを設定します。
    # kwargs["adapter"] = HTTPAdapter()

    # 証明書検証をスキップするかどうかを指定します。デフォルト：false。
    # kwargs["insecure_skip_verify"] = False

    # HTTP リダイレクトを有効にするかどうかを指定します。デフォルト：false。
    # 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

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...


# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

アクセス認証情報の設定

OSS は、認証情報を初期化するための複数の方法を提供します。認証と権限付与の要件に基づいて方法を選択できます。

RAM ユーザーの AccessKey ペアの使用

Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey ペア (AccessKey ID と AccessKey Secret) を使用して、認証情報プロバイダーを初期化できます。この方法は、ご利用のアプリケーションが外部攻撃を受けにくい安全で安定した環境で実行され、OSS への長期アクセスを必要とし、認証情報を頻繁にローテーションできない場合に適しています。ただし、この方法では AccessKey ペアを手動で維持する必要があり、セキュリティリスクとメンテナンスの複雑さが増します。

import alibabacloud_oss_v2 as oss

def main():
    # 静的認証情報プロバイダーを作成し、AccessKey ID と AccessKey Secret を明示的に設定します。ご利用の 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

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...


# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

一時的な STS トークンの使用

ご利用のアプリケーションが OSS への一時的なアクセスを必要とする場合、Security Token Service (STS) から取得した一時的な ID 認証情報 (AccessKey ID、AccessKey Secret、およびセキュリティトークン) を使用して認証情報プロバイダーを初期化できます。ただし、この方法では STS トークンを手動で維持する必要があり、セキュリティリスクとメンテナンスの複雑さが増します。さらに、OSS に複数回一時的にアクセスするには、STS トークンを手動でリフレッシュする必要があります。

import alibabacloud_oss_v2 as oss

def main():
    # Alibaba Cloud アカウントの AccessKey ID と AccessKey Secret ではなく、取得した一時的な AccessKey ID と AccessKey Secret を指定します。
    # STS から取得した AccessKey ID は、以下に示すように「STS」で始まることに注意してください。
    sts_access_key_id = 'STS.****************'
    sts_access_key_secret = 'yourAccessKeySecret'
    # 取得した STS セキュリティトークンを指定します。
    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

    # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
    cfg.region = 'cn-hangzhou'

    # 指定された設定で OSS クライアントを作成します。
    client = oss.Client(cfg)

    # 作成されたクライアントを使用して後続の操作を実行します...


# このスクリプトが直接実行される場合は、main 関数を呼び出します。
if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

RAM ロール ARN の使用

アプリケーションがクロスアカウントアクセスなど、OSS への承認されたアクセスを必要とする場合、RAM ロール Alibaba Cloud リソースネーム (ARN) を使用して認証情報プロバイダーを初期化できます。このメソッドは STS トークンに基づいています。RAM ロールの ARN を指定すると、認証情報ツールは STS から STS トークンを取得します。その後、ツールは現在のトークンの有効期限が切れる前に AssumeRole 操作を呼び出して、新しい STS トークンをリクエストします。また、policy パラメーターに値を割り当てて、RAM ロールをより小さい権限のセットに制限することもできます。

1. alibabacloud_credentials 依存関係を追加します。

   pip install alibabacloud_credentials

2. AccessKey ペアと RAM ロール ARN をアクセス認証情報として設定します。

   # -*- 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 ペア (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
   
       # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
       cfg.region = 'cn-hangzhou'
   
       # OSS クライアントインスタンスを作成します。
       client = oss.Client(cfg)
   
       # クライアントを使用して後続の操作を実行します...
   
   
   # このスクリプトが直接実行される場合は、main 関数を呼び出します。
   if __name__ == "__main__":
       main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

ECS RAM ロールの使用

ご利用のアプリケーションが ECS インスタンス、ECI インスタンス、または Container Service for Kubernetes (ACK) のワーカーノードで実行される場合、ECS RAM ロールを使用して認証情報プロバイダーを初期化することを推奨します。この方法は STS トークンに基づいています。ECS RAM ロールを使用すると、ロールを ECS インスタンス、ECI インスタンス、または ACK のワーカーノードに関連付けて、インスタンス内で STS トークンを自動的にリフレッシュできます。この方法では、AccessKey ペアや STS トークンを提供する必要がなく、手動メンテナンスのリスクを軽減できます。ECS RAM ロールの取得方法の詳細については、「CreateRole - RAM ロールの作成」をご参照ください。

1. alibabacloud_credentials 依存関係を追加します。

   pip install alibabacloud_credentials

2. ECS RAM ロールをアクセス認証情報として設定します。

   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
   
       # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
       cfg.region = 'cn-hangzhou'
   
       # OSS クライアントインスタンスを作成します。
       client = oss.Client(cfg)
   
       # クライアントを使用して後続の操作を実行します...
   
   
   # このスクリプトが直接実行される場合は、main 関数を呼び出します。
   if __name__ == "__main__":
       main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

OIDC ロール ARN の使用

ACK でワーカーノードの RAM ロールを設定すると、そのノード上の Pod 内のアプリケーションは、グローバルメタサービスを通じて関連付けられたロールの STS トークンを取得できます。これは、ECS にデプロイされたアプリケーションがトークンを取得する方法と似ています。ただし、コンテナークラスター上の信頼できないアプリケーションがワーカーノードのインスタンス RAM ロールの STS トークンを取得することを望まない場合があります。信頼できないアプリケーションの例としては、コードが公開されていない顧客から提出されたアプリケーションなどがあります。セキュリティリスクを防ぎ、これらの信頼できないアプリケーションが必要な STS トークンを最小限の権限で安全に取得できるようにするには、RAM Roles for Service Accounts (RRSA) 機能を使用できます。この方法は STS トークンに基づいています。Alibaba Cloud コンテナークラスターは、さまざまなアプリケーション Pod に対応するサービスアカウントの OpenID Connect (OIDC) トークンファイルを作成してマウントし、関連する設定情報を環境変数に注入します。認証情報ツールは、環境変数から設定情報を取得し、STS の AssumeRoleWithOIDC 操作を呼び出して、その情報をバインドされたロールの STS トークンと交換します。この方法では、AccessKey ペアや STS トークンを提供する必要がなく、手動メンテナンスのリスクを軽減できます。詳細については、「RRSA に基づいて 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
       
       # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
       cfg.region = "cn-hangzhou"
   
       # 指定された設定で OSS クライアントを作成します。
       client = oss.Client(cfg)
   
       # クライアントを使用して後続の操作を実行します...
   
   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_secret。
           return oss.credentials.Credentials('<access_key_id>', '<access_key_secrect>')
   
           # 一時的な STS 認証情報を返します：access_key_id, access_key_secret, 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
   
       # バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
       cfg.region = 'cn-hangzhou'
   
       client = oss.Client(cfg)
   
       # クライアントを使用して後続の操作を実行します...
   
   
   if __name__ == "__main__":
       main()  # スクリプトのエントリポイント。ファイルが直接実行されると main 関数を呼び出します。

匿名アクセス

公開読み取り可能な OSS リソースにのみアクセスする必要がある場合は、認証情報を提供せずに匿名アクセスを使用できます。

import alibabacloud_oss_v2 as oss

# 匿名認証情報プロバイダーを作成します。
credentials_provider = oss.credentials.AnonymousCredentialsProvider()

# クライアントを設定します。
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
# バケットが配置されているリージョンを指定します。たとえば、中国 (杭州) のリージョン ID は cn-hangzhou です。
cfg.region = 'cn-hangzhou'

# OSS クライアントを作成します。
client = oss.Client(cfg)

セルフサービスのトラブルシューティング

OSS Python SDK V2 を使用して OSS にアクセスする際にエラーが発生した場合、OSS は HTTP ステータスコード、メッセージ、リクエスト ID、EC エラーコードなどの情報を返します。EC エラーコードはエラーの具体的な原因を示します。EC エラーコードを使用してセルフサービスでトラブルシューティングを行うことができます。

コード例

OSS Python SDK V2 は、ご参照いただくためにさまざまなコード例を提供しています。