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

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

注意事項

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

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

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

マルチパートアップロードを初期化する。
Client.InitiateMultipartUpload メソッドを呼び出して、OSS に固有のアップロード ID を取得します。
パーツをアップロードする。
Client.UploadPart メソッドを呼び出して、パーツデータをアップロードします。
説明
- 特定のアップロード ID について、パーツ番号は各パーツと最終オブジェクト内の相対位置を識別します。既存のパーツ番号を使用して新しいパーツをアップロードすると、元のパーツは上書きされます。
- OSS は、レスポンスの ETag ヘッダーにパーツデータの MD5 ハッシュを含めます。
- OSS は、アップロードされたデータの MD5 ハッシュを計算し、SDK によって計算されたハッシュと比較します。
マルチパートアップロードを完了する。
Client.CompleteMultipartUpload メソッドを呼び出して、これらのパーツを完全なオブジェクトに結合します。

サンプルコード

次のサンプルコードは、大きなローカルファイルを複数の部分に分割し、パーツをバケットにアップロードしてから、これらのパーツを完全なオブジェクトに結合する方法を示しています。

import os
import argparse
import alibabacloud_oss_v2 as oss

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

# 必須の --region パラメータ。バケットが配置されているリージョンを指定します。
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 必須の --bucket パラメータ。バケットの名前を指定します。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# オプションの --endpoint パラメータ。他のサービスが OSS にアクセスするために使用できるエンドポイントを指定します。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 必須の --key パラメータ。オブジェクトの名前を指定します。
parser.add_argument('--key', help='The name of the object.', required=True)

# 必須の --file_path パラメータ。アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='The path of Upload file.', 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

    # 構成を使用して OSSClient インスタンスを作成します。
    client = oss.Client(cfg)

    # マルチパートアップロードを開始し、アップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パーツのサイズを 5MB として定義します。
    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)

    # マルチパートアップロードを完了し、すべてのパーツを完全なオブジェクトに結合するためのリクエストを送信します。
    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
        )
    ))

    # 次のコードは、サーバー側のリストメソッドを使用してすべてのパーツデータを完全なオブジェクトにマージする別の方法です。
    # このメソッドは、すべてのパーツが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側のリストメソッドを介して断片化されたデータを完全なオブジェクトにマージします
    # 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='The region in which the bucket is located.', required=True)

# 必須の --bucket パラメータ。バケットの名前を指定します。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# オプションの --endpoint パラメータ。他のサービスが OSS にアクセスするために使用できるエンドポイントを指定します。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 必須の --key パラメータ。オブジェクトの名前を指定します。
parser.add_argument('--key', help='The name of the object.', required=True)

# 必須の --file_path パラメータ。アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='The path of Upload file.', 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

    # 構成を使用して OSSClient インスタンスを作成します。
    client = oss.Client(cfg)

    # マルチパートアップロードを開始し、アップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パーツのサイズを 1MB として定義します。
    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)

    # コールバック URL を定義します。
    call_back_url = "http://www.example.com/callback"
    # コールバックパラメータを構築します。コールバック URL とリクエスト本文を 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()
    # Base64 エンコーディングを使用してカスタム変数を構築します
    callback_var=base64.b64encode('{\"x:var1\":\"value1\",\"x:var2\":\"value2\"}'.encode()).decode()

    # マルチパートアップロードを完了し、すべてのパーツを完全なオブジェクトに結合するためのリクエストを送信します。
    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
    ))

    # 次のコードは、サーバー側のリストメソッドを使用してすべてのパーツデータを完全なオブジェクトにマージする別の方法です。
    # このメソッドは、すべてのパーツが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側のリストメソッドを介して断片化されたデータを完全なオブジェクトにマージします
    # 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='The region in which the bucket is located.', required=True)

# 必須の --bucket パラメータ。バケットの名前を指定します。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# オプションの --endpoint パラメータ。他のサービスが OSS にアクセスするために使用できるエンドポイントを指定します。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 必須の --key パラメータ。オブジェクトの名前を指定します。
parser.add_argument('--key', help='The name of the object.', required=True)

# 必須の --file_path パラメータ。アップロードするファイルのパスを指定します。
parser.add_argument('--file_path', help='The path of Upload file.', 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

    # endpoint パラメータが提供されている場合は、エンドポイントを指定します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 構成を使用して OSSClient インスタンスを作成します。
    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 が現在の行を上書きできるようにします。
        print(f'\rUpload progress: {rate}% ', end='')

    # マルチパートアップロードを開始し、アップロード ID を取得します。
    result = client.initiate_multipart_upload(oss.InitiateMultipartUploadRequest(
        bucket=args.bucket,
        key=args.key,
    ))

    # 各パーツのサイズを 5MB として定義します。
    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)

    # マルチパートアップロードを完了し、すべてのパーツを完全なオブジェクトに結合するためのリクエストを送信します。
    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
        )
    ))

    # 次のコードは、サーバー側のリストメソッドを使用してすべてのパーツデータを完全なオブジェクトにマージする別の方法です。
    # このメソッドは、すべてのパーツが正常にアップロードされたかどうか不明な場合に適しています。
    # サーバー側のリストメソッドを介して断片化されたデータを完全なオブジェクトにマージします
    # 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 関数が呼び出されます。

参照

マルチパートアップロードの完全なサンプルコードについては、「complete_multipart_upload.py」を参照してください。