Object Storage Service:範囲のダウンロード (Python SDK V2)

使用上の注意

• このトピックのサンプルコードでは、中国 (杭州) リージョンのリージョン ID cn-hangzhou を使用します。デフォルトでは、パブリックエンドポイントを使用してバケット内のリソースにアクセスします。同じリージョン内の他の Alibaba Cloud サービスからバケット内のリソースにアクセスする場合は、内部エンドポイントを使用します。OSS がサポートするリージョンとエンドポイントの詳細については、「OSS のリージョンとエンドポイント」をご参照ください。

• 範囲のダウンロードを使用するには、oss:GetObject 権限が必要です。詳細については、「RAM ユーザーへのカスタムポリシーの付与」をご参照ください。

メソッド定義

get_object(request: GetObjectRequest, **kwargs) → GetObjectResult

リクエストパラメーター

応答パラメーター

サンプルコード

次のコードは、リクエストに RangeBehavior:standard リクエストヘッダーを追加する方法を示しています。このヘッダーは、指定された範囲内のファイルデータをダウンロードする標準のダウンロード動作を指定します。

import argparse
import alibabacloud_oss_v2 as oss

# コマンドラインパラメーターパーサーを作成します。
parser = argparse.ArgumentParser(description="Get object range sample")
# 必要なコマンドラインパラメーターを追加します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)
parser.add_argument('--bucket', help='バケットの名前。', required=True)
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')
parser.add_argument('--key', help='オブジェクトの名前。', required=True)
parser.add_argument('--range', help='ファイル転送の範囲を指定します。値の例: bytes=0-9', required=True)
parser.add_argument('--range_behavior', help='指定された範囲内のダウンロード動作の標準。値の例: standard。')

def main():
    # コマンドラインパラメーターを解析します。
    args = parser.parse_args()

    # 環境変数からアクセス資格情報を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # SDK のデフォルト構成を使用します。
    cfg = oss.config.load_default()
    # 資格情報プロバイダーを指定します。
    cfg.credentials_provider = credentials_provider
    # リージョンを指定します。
    cfg.region = args.region
    # コマンドラインからエンドポイントが提供された場合、構成内のエンドポイントを提供されたエンドポイントで更新します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

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

    # オブジェクトを取得するリクエストを開始します。
    result = client.get_object(oss.GetObjectRequest(
        bucket=args.bucket,  # バケットの名前を指定します
        key=args.key,  # オブジェクトのキー名を指定します
        range_header=args.range,  # 範囲ヘッダーを指定します
        range_behavior=args.range_behavior,  # 範囲内のダウンロード動作を指定します
    ))

    # 応答の複数の属性を出力します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content length: {result.content_length},'
          f' content range: {result.content_range},'
          f' content type: {result.content_type},'
          f' etag: {result.etag},'
          f' last modified: {result.last_modified},'
          f' content md5: {result.content_md5},'
          f' cache control: {result.cache_control},'
          f' content disposition: {result.content_disposition},'
          f' content encoding: {result.content_encoding},'
          f' expires: {result.expires},'
          f' hash crc64: {result.hash_crc64},'
          f' storage class: {result.storage_class},'
          f' object type: {result.object_type},'
          f' version id: {result.version_id},'
          f' tagging count: {result.tagging_count},'
          f' server side encryption: {result.server_side_encryption},'
          f' server side data encryption: {result.server_side_data_encryption},'
          f' next append position: {result.next_append_position},'
          f' expiration: {result.expiration},'
          f' restore: {result.restore},'
          f' process status: {result.process_status},'
          f' delete marker: {result.delete_marker},'
    )

    # ========== 方法 1: 完全に読み取る ==========
    # コンテキストマネージャーを使用してリソースの解放を保証します。
    with result.body as body_stream:
        data = body_stream.read()
        print(f"ファイルの読み取りが完了しました、データ長: {len(data)} バイト")

        path = "./get-object-sample.txt"
        with open(path, 'wb') as f:
            f.write(data)
        print(f"ファイルのダウンロードが完了しました、保存先パス: {path}")

    # # ========== 方法 2: チャンクで読み取る ==========
    # # コンテキストマネージャーを使用してリソースの解放を保証します。
    # with result.body as body_stream:
    #     chunk_path = "./get-object-sample-chunks.txt"
    #     total_size = 0

    #     with open(chunk_path, 'wb') as f:
    #         # 256 KB のブロックサイズを使用します (必要に応じて block_size パラメーターを調整します)。
    #         for chunk in body_stream.iter_bytes(block_size=256 * 1024):
    #             f.write(chunk)
    #             total_size += len(chunk)
    #             print(f"データブロックを受信しました: {len(chunk)} バイト | 合計: {total_size} バイト")

    #     print(f"ファイルのダウンロードが完了しました、保存先パス: {chunk_path}")

# スクリプトがメインプログラムとして実行されるときに main 関数を呼び出します。
if __name__ == "__main__":
    main()

リファレンス

• 範囲のダウンロードの完全なサンプルコードについては、「get_object_range.py」をご参照ください。