OSS SDK for Python V2 を使用したマルチパートアップロードの実行方法 - Object Storage Service

Object Storage Service (OSS) のマルチパートアップロード機能を使用すると、大きなオブジェクトを複数のパートに分割できます。これらのパートをアップロードした後、CompleteMultipartUpload API を呼び出してパートを結合し、完全なオブジェクトを作成できます。

注意事項

このトピックのサンプルコードは、デフォルトで中国 (杭州) リージョン (cn-hangzhou) とそのパブリックエンドポイントを使用します。同じリージョン内の他の Alibaba Cloud プロダクトから OSS にアクセスする場合は、内部エンドポイントを使用してください。 OSS でサポートされているリージョンとエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。
マルチパートアップロードを実行するには、oss:PutObject 権限が必要です。詳細については、「RAM ユーザーにカスタム権限を付与する」をご参照ください。

マルチパートアップロードのプロセス

マルチパートアップロードは、次の 3 つのステップで構成されます。

マルチパートアップロードイベントを初期化します。

Client.InitiateMultipartUpload メソッドを呼び出して、OSS からグローバルに一意なアップロード ID を取得します。
パートをアップロードします。

Client.UploadPart メソッドを呼び出して、パートデータをアップロードします。
説明
- 同じアップロード ID の場合、パート番号はパートと、完全なオブジェクト内でのその相対位置を示します。既存のパート番号で新しいパートをアップロードすると、OSS 上の既存のパートが上書きされます。
- OSS は、受信したパートデータの MD5 ハッシュをレスポンスの ETag ヘッダーに含めます。
- OSS は、アップロードされたデータの MD5 ハッシュを計算し、ソフトウェア開発キット (SDK) によって計算された MD5 ハッシュと比較します。2 つの MD5 ハッシュが異なる場合、InvalidDigest エラーコードが返されます。
マルチパートアップロードを完了します。

すべてのパートがアップロードされたら、Client.CompleteMultipartUpload メソッドを呼び出して、すべてのパートを 1 つの完全なオブジェクトにマージできます。

サンプルコード

次のサンプルコードは、大きなローカルファイルを複数のパートに分割し、パートをバケットに並行してアップロードした後、パートをマージして完全なオブジェクトを作成する方法を示しています。

import os
import argparse
import alibabacloud_oss_v2 as oss

# マルチパートアップロードサンプル用のコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="multipart upload sample")

# 必須の --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)

# 必須の --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。
parser.add_argument('--bucket', help='バケットの名前。', required=True)

# オプションの --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')

# 必須の --key コマンドライン引数を追加します。これは、オブジェクトの名前を指定します。
parser.add_argument('--key', help='オブジェクトの名前。', required=True)

# 必須の --file_path コマンドライン引数を追加します。これは、アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='アップロードするファイルのパス。', required=True)


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)

    # マルチパートアップロードリクエストを開始して、後続のパートアップロードに使用するアップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パートのサイズを 5 MB として定義します。
    part_size = 5 * 1024 * 1024

    # アップロードするファイルの合計サイズを取得します。
    data_size = os.path.getsize(args.file_path)

    # パート番号を 1 から初期化します。
    part_number = 1

    # 各パートアップロードの結果を保存します。
    upload_parts = []

    # ファイルをバイナリ読み取りモードで開きます。
    with open(args.file_path, 'rb') as f:
        # part_size に基づいてファイルを走査し、パートごとにアップロードします。
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # 最後のパートが part_size より小さい場合を処理します。
                n = data_size - start

            # ファイルの特定の部分を読み取るための SectionReader を作成します。
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # パートをアップロードします。
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader
            ))

            # 各パートアップロードの結果を出力します。
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # パートアップロード結果をリストに保存します。
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # パート番号をインクリメントします。
            part_number += 1

    # アップロードされたパートをパート番号でソートします。
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # マルチパートアップロードを完了するリクエストを送信し、すべてのパートを 1 つの完全なオブジェクトにマージします。
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        )
    ))

    # 次のコードは、サーバー上のすべてのパートデータをリストしてマージし、完全なオブジェクトを作成する別の方法を示しています。
    # この方法は、すべてのパートが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側の List メソッドを使用して、断片化されたデータを完全なオブジェクトにマージします。
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # 完了したマルチパートアップロードの結果を出力します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されたときに main 関数が呼び出されます。

一般的なシナリオ

マルチパートアップロードを実行してアップロードコールバックを設定する

マルチパートアップロードが完了した後にアプリケーションサーバーに通知する場合は、次のサンプルコードをご参照ください。

import os
import argparse
import base64
import alibabacloud_oss_v2 as oss

# マルチパートアップロードサンプル用のコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="multipart upload sample")

# 必須の --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)

# 必須の --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。
parser.add_argument('--bucket', help='バケットの名前。', required=True)

# オプションの --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')

# 必須の --key コマンドライン引数を追加します。これは、オブジェクトの名前を指定します。
parser.add_argument('--key', help='オブジェクトの名前。', required=True)

# 必須の --file_path コマンドライン引数を追加します。これは、アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='アップロードするファイルのパス。', required=True)


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)

    # マルチパートアップロードリクエストを開始して、後続のパートアップロードに使用するアップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パートのサイズを 1 MB として定義します。
    part_size = 1 * 1024 * 1024

    # アップロードするファイルの合計サイズを取得します。
    data_size = os.path.getsize(args.file_path)

    # パート番号を 1 から初期化します。
    part_number = 1

    # 各パートアップロードの結果を保存します。
    upload_parts = []

    # ファイルをバイナリ読み取りモードで開きます。
    with open(args.file_path, 'rb') as f:
        # part_size に基づいてファイルを走査し、パートごとにアップロードします。
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # 最後のパートが part_size より小さい場合を処理します。
                n = data_size - start

            # ファイルの特定の部分を読み取るための SectionReader を作成します。
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # パートをアップロードします。
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader
            ))

            # 各パートアップロードの結果を出力します。
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # パートアップロード結果をリストに保存します。
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # パート番号をインクリメントします。
            part_number += 1

    # アップロードされたパートをパート番号でソートします。
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # Webhook アドレスを定義します。
    call_back_url = "http://www.example.com/callback"
    # コールバックパラメータを作成します。Webhook アドレスとリクエスト本文を指定し、Base64 でエンコードします。
    callback=base64.b64encode(str('{\"callbackUrl\":\"' + call_back_url + '\",\"callbackBody\":\"bucket=${bucket}&object=${object}&my_var_1=${x:var1}&my_var_2=${x:var2}\"}').encode()).decode()
    # カスタム変数 (callback-var) を作成し、Base64 でエンコードします。
    callback_var=base64.b64encode('{\"x:var1\":\"value1\",\"x:var2\":\"value2\"}'.encode()).decode()

    # マルチパートアップロードを完了するリクエストを送信し、すべてのパートを 1 つの完全なオブジェクトにマージします。
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        ),
        callback=callback,
        callback_var=callback_var
    ))

    # 次のコードは、サーバー上のすべてのパートデータをリストしてマージし、完全なオブジェクトを作成する別の方法を示しています。
    # この方法は、すべてのパートが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側の List メソッドを使用して、断片化されたデータを完全なオブジェクトにマージします。
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # 完了したマルチパートアップロードの結果を出力します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されたときに main 関数が呼び出されます。

マルチパートアップロードのプログレスバーを表示する

import os
import argparse
import alibabacloud_oss_v2 as oss

# マルチパートアップロードサンプル用のコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="multipart upload sample")

# 必須の --region コマンドライン引数を追加します。これは、バケットが配置されているリージョンを指定します。
parser.add_argument('--region', help='バケットが配置されているリージョン。', required=True)

# 必須の --bucket コマンドライン引数を追加します。これは、バケットの名前を指定します。
parser.add_argument('--bucket', help='バケットの名前。', required=True)

# オプションの --endpoint コマンドライン引数を追加します。これは、他のサービスが OSS にアクセスするために使用できるドメイン名を指定します。
parser.add_argument('--endpoint', help='他のサービスが OSS にアクセスするために使用できるドメイン名')

# 必須の --key コマンドライン引数を追加します。これは、オブジェクトの名前を指定します。
parser.add_argument('--key', help='オブジェクトの名前。', required=True)

# 必須の --file_path コマンドライン引数を追加します。これは、アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='アップロードするファイルのパス。', required=True)

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)

    # progress_state という名前の辞書変数を定義して、アップロードの進行状況を保存します。初期値は 0 です。
    progress_state = {'saved': 0}
    def _progress_fn(n, written, total):
        # グローバル変数の使用を避けるために、辞書を使用して累積書き込みバイト数を保存します。
        progress_state['saved'] += n

        # 現在のアップロード進捗率を計算します。書き込まれたバイト数を合計バイト数で除算し、結果を切り捨てます。
        rate = int(100 * (float(written) / float(total)))

        # 現在のアップロード進行状況を出力します。\r はカーソルを行の先頭に戻し、コマンドラインでのリアルタイム更新を可能にします。
        # end='' は改行を防ぎ、次の出力で現在の行を上書きできるようにします。
        print(f'\rアップロード進行状況: {rate}% ', end='')

    # マルチパートアップロードリクエストを開始して、後続のパートアップロードに使用するアップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パートのサイズを 5 MB として定義します。
    part_size = 5 * 1024 * 1024

    # アップロードするファイルの合計サイズを取得します。
    data_size = os.path.getsize(args.file_path)

    # パート番号を 1 から初期化します。
    part_number = 1

    # 各パートアップロードの結果を保存します。
    upload_parts = []

    # ファイルをバイナリ読み取りモードで開きます。
    with open(args.file_path, 'rb') as f:
        # part_size に基づいてファイルを走査し、パートごとにアップロードします。
        for start in range(0, data_size, part_size):
            n = part_size
            if start + n > data_size:  # 最後のパートが part_size より小さい場合を処理します。
                n = data_size - start

            # ファイルの特定の部分を読み取るための SectionReader を作成します。
            reader = oss.io_utils.SectionReader(oss.io_utils.ReadAtReader(f), start, n)

            # パートをアップロードします。
            up_result = client.upload_part(oss.UploadPartRequest(
                bucket=args.bucket,
                key=args.key,
                upload_id=result.upload_id,
                part_number=part_number,
                body=reader,
                progress_fn=_progress_fn
            ))

            # 各パートアップロードの結果を出力します。
            print(f'status code: {up_result.status_code},'
                  f' request id: {up_result.request_id},'
                  f' part number: {part_number},'
                  f' content md5: {up_result.content_md5},'
                  f' etag: {up_result.etag},'
                  f' hash crc64: {up_result.hash_crc64},'
                  )

            # パートアップロード結果をリストに保存します。
            upload_parts.append(oss.UploadPart(part_number=part_number, etag=up_result.etag))

            # パート番号をインクリメントします。
            part_number += 1

    # アップロードされたパートをパート番号でソートします。
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # マルチパートアップロードを完了するリクエストを送信し、すべてのパートを 1 つの完全なオブジェクトにマージします。
    result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
        upload_id=result.upload_id,
        complete_multipart_upload=oss.CompleteMultipartUpload(
            parts=parts
        )
    ))

    # 次のコードは、サーバー上のすべてのパートデータをリストしてマージし、完全なオブジェクトを作成する別の方法を示しています。
    # この方法は、すべてのパートが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側の List メソッドを使用して、断片化されたデータを完全なオブジェクトにマージします。
    # result = client.complete_multipart_upload(oss.CompleteMultipartUploadRequest(
    #     bucket=args.bucket,
    #     key=args.key,
    #     upload_id=result.upload_id,
    #     complete_all='yes'
    # ))

    # 完了したマルチパートアップロードの結果を出力します。
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' bucket: {result.bucket},'
          f' key: {result.key},'
          f' location: {result.location},'
          f' etag: {result.etag},'
          f' encoding type: {result.encoding_type},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # スクリプトのエントリポイント。ファイルが直接実行されたときに main 関数が呼び出されます。

Object Storage Service:マルチパートアップロード (Python SDK V2)