OSS SDK for Python を使用してマルチパートアップロードを実行し、ラージオブジェクトをアップロードする - Object Storage Service

5 GB を超えるサイズのラージオブジェクトを Object Storage Service (OSS) にアップロードすると、ネットワークの中断またはプログラムのクラッシュが原因で、オブジェクトのアップロードに失敗することがあります。複数回再試行してもオブジェクトのアップロードに失敗する場合は、マルチパートアップロードを実行することでラージオブジェクトをアップロードできます。オブジェクトを複数のパートに分割し、パートを並行してアップロードすることで、アップロードを高速化できます。すべてのパートがアップロードされた後、CompleteMultipartUpload 操作を呼び出して、これらのパートを完全なオブジェクトに結合できます。

使用上の注意

このトピックでは、中国 (杭州) リージョンのパブリックエンドポイントが使用されています。OSS と同じリージョン内の他の Alibaba Cloud サービスから OSS にアクセスする場合は、内部エンドポイントを使用します。OSS のリージョンとエンドポイントの詳細については、「リージョンとエンドポイント」をご参照ください。
このトピックでは、アクセス認証情報は環境変数から取得されます。アクセス認証情報を設定する方法の詳細については、「アクセス認証情報を設定する (Python SDK V1)」をご参照ください。
このトピックでは、OSS エンドポイントを使用して OSSClient インスタンスが作成されます。カスタムドメイン名または Security Token Service (STS) を使用して OSSClient インスタンスを作成する場合は、「初期化 (Python SDK V1)」をご参照ください。
InitiateMultipartUpload、UploadPart、CompleteMultipartUpload を含むマルチパートアップロードの完全なワークフローを完了するには、oss:PutObject 権限が必要です。詳細については、「RAM ユーザーにカスタムポリシーをアタッチする」をご参照ください。

プロセス

マルチパートアップロードを実装するには、次の手順を実行します。

マルチパートアップロードタスクを開始します。
bucket.init_multipart_upload メソッドを使用して、OSS で一意のアップロード ID を取得します。
パートをアップロードします。
bucket.upload_part メソッドを使用して、パートをアップロードします。
説明
- マルチパートアップロードタスクでは、パート番号を使用して、オブジェクト内のパートの相対位置を識別します。パートをアップロードし、そのパート番号を使用して別のパートをアップロードすると、後者のパートは前者のパートを上書きします。
- OSS は、アップロードされた各パートの MD5 ハッシュを、レスポンスの ETag ヘッダーに含めます。
- OSS は、アップロードされたパートの MD5 ハッシュを計算し、その MD5 ハッシュを OSS SDK for Go によって計算された MD5 ハッシュと比較します。2 つのハッシュが異なる場合、OSS は InvalidDigest エラーコードを返します。
マルチパートアップロードタスクを完了します。
すべてのパートをアップロードした後、bucket.complete_multipart_upload メソッドを使用して、パートを完全なオブジェクトに結合します。

例

すべてのパートがアップロードされた後、次のいずれかの方法を使用して、すべてのパートを完全なオブジェクトに結合できます。

リクエスト本文にパート情報を含めることによって、すべてのパートを完全なオブジェクトに結合します。

# -*- coding: utf-8 -*-
import os
from oss2 import SizedFileAdapter, determine_part_size
from oss2.models import PartInfo
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 環境変数からアクセス認証情報を取得します。サンプルコードを実行する前に、OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# オブジェクトの完全なパスを指定します。完全なパスにバケット名を含めないでください。例: exampledir/exampleobject.txt。
key = 'exampledir/exampleobject.txt'
# アップロードするローカルファイルの完全なパスを指定します。例: D:\\localpath\\examplefile.txt。
filename = 'D:\\localpath\\examplefile.txt'

total_size = os.path.getsize(filename)
# determine_part_size メソッドを使用して、パートサイズを決定します。最小パートサイズは 100 KB、最大パートサイズは 5 GB です。最後のパートは 100 KB より小さくできます。この例では、パートサイズは 1 MB に設定されています。
part_size = determine_part_size(total_size, preferred_size=1 * 1024 * 1024)

# マルチパートアップロードタスクを開始します。
# マルチパートアップロードタスクを開始するときにオブジェクトのストレージタイプを指定する場合は、init_multipart_upload メソッドを使用するときに関連するヘッダーを設定します。
# headers = dict()
# ウェブページのオブジェクトのキャッシュ動作を指定します。
# headers['Cache-Control'] = 'no-cache'
# ダウンロード時のオブジェクトの名前を指定します。
# headers['Content-Disposition'] = 'oss_MultipartUpload.txt'
# 有効期間を指定します。単位: ミリ秒。
# headers['Expires'] = '1000'
# マルチパートアップロードを実行してアップロードされたオブジェクトが、マルチパートアップロードタスクの開始時に同じ名前の既存のオブジェクトを上書きするかどうかを指定します。この例では、このパラメーターは true に設定されており、同じ名前のオブジェクトを上書きできないことを示しています。
# headers['x-oss-forbid-overwrite'] = 'true'
# 各パートの暗号化に使用するサーバー側暗号化方式を指定します。
# headers[OSS_SERVER_SIDE_ENCRYPTION] = SERVER_SIDE_ENCRYPTION_KMS
# オブジェクトの暗号化に使用するアルゴリズムを指定します。このパラメーターを設定しない場合、オブジェクトは AES-256 を使用して暗号化されます。
# headers[OSS_SERVER_SIDE_DATA_ENCRYPTION] = SERVER_SIDE_ENCRYPTION_KMS
# Key Management Service (KMS) によって管理される Customer Master Key (CMK) の ID を指定します。
# headers[OSS_SERVER_SIDE_ENCRYPTION_KEY_ID] = '9468da86-3509-4f8d-a61e-6eab1eac****'
# オブジェクトのストレージタイプを指定します。
# headers['x-oss-storage-class'] = oss2.BUCKET_STORAGE_CLASS_STANDARD
# オブジェクトのタグを指定します。オブジェクトに複数のタグを同時に指定できます。
# headers[OSS_OBJECT_TAGGING] = 'k1=v1&k2=v2&k3=v3'
# upload_id = bucket.init_multipart_upload(key, headers=headers).upload_id
upload_id = bucket.init_multipart_upload(key).upload_id
# アップロード ID に基づいてマルチパートアップロードタスクをキャンセルするか、アップロードされたパートをリストします。
# アップロード ID に基づいてマルチパートアップロードタスクをキャンセルする場合は、InitiateMultipartUpload 操作を呼び出してマルチパートアップロードタスクを開始した後に、アップロード ID を取得します。
# アップロード ID に基づいてマルチパートアップロードタスクでアップロードされたパートをリストする場合は、InitiateMultipartUpload 操作を呼び出してマルチパートアップロードタスクを開始した後、CompleteMultipartUpload 操作を呼び出してマルチパートアップロードタスクを完了する前に、アップロード ID を取得します。
# print("UploadID:", upload_id)
parts = []

# パートをアップロードします。
with open(filename, 'rb') as fileobj:
    part_number = 1
    offset = 0
    while offset < total_size:
        num_to_upload = min(part_size, total_size - offset)
        # SizedFileAdapter(fileobj, size) メソッドを使用して新しいオブジェクトを生成し、追加操作の開始位置を再計算します。
        result = bucket.upload_part(key, upload_id, part_number,
                                    SizedFileAdapter(fileobj, num_to_upload))
        parts.append(PartInfo(part_number, result.etag))

        offset += num_to_upload
        part_number += 1

# マルチパートアップロードタスクを完了します。
# マルチパートアップロードタスクを完了するときにヘッダーを設定します (必要な場合)。
headers = dict()
# オブジェクトのアクセス制御リスト (ACL) を指定します。この例では、ACL は OBJECT_ACL_PRIVATE に設定されており、オブジェクトの ACL が非公開であることを示しています。
# headers["x-oss-object-acl"] = oss2.OBJECT_ACL_PRIVATE
bucket.complete_multipart_upload(key, upload_id, parts, headers=headers)
# bucket.complete_multipart_upload(key, upload_id, parts)

重要

ネットワークの状態が安定している場合は、各パートのサイズを大きくすることをお勧めします。そうでない場合は、各パートのサイズを小さくします。

サーバー上のパートをリストすることによって、パートを完全なオブジェクトに結合します。

説明

サーバー上のパートをリストすることによってパートを完全なオブジェクトに結合する場合は、次のサンプルコードで指定されたアップロード ID を使用して複数のパートがアップロードされていることを確認してください。

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# 環境変数からアクセス認証情報を取得します。サンプルコードを実行する前に、OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# オブジェクトの完全なパスを指定します。完全なパスにバケット名を含めないでください。例: exampledir/exampleobject.txt。
key = 'exampledir/exampleobject.txt'
# アップロードするローカルファイルの完全なパスを指定します。例: D:\\localpath\\examplefile.txt。
filename = 'D:\\localpath\\examplefile.txt'
# アップロード ID を指定します。アップロード ID は、InitiateMultipartUpload 操作を呼び出してマルチパートアップロードタスクを開始した後、CompleteMultipartUpload 操作を呼び出してマルチパートアップロードタスクを完了する前に取得できます。
upload_id = '0004B9894A22E5B1888A1E29F823****'

# マルチパートアップロードタスクを完了します。
# マルチパートアップロードタスクを完了するときにオブジェクトの ACL を指定する場合は、complete_multipart_upload 関数で関連するヘッダーを設定します。
headers = dict()
# headers["x-oss-object-acl"] = oss2.OBJECT_ACL_PRIVATE
# リクエストで x-oss-complete-all:yes を指定すると、OSS は現在のアップロード ID を使用してアップロードされたすべてのパートをリストし、パート番号でパートをソートしてから、CompleteMultipartUpload 操作を実行します。
# リクエストで x-oss-complete-all:yes が指定されている場合、リクエスト本文を指定することはできません。指定すると、エラーが発生します。
headers["x-oss-complete-all"] = 'yes'
bucket.complete_multipart_upload(key, upload_id, None, headers=headers)

マルチパートアップロードタスクをキャンセルする

bucket.abort_multipart_upload メソッドを使用して、マルチパートアップロードタスクをキャンセルできます。マルチパートアップロードタスクがキャンセルされると、アップロード ID を使用してパートをアップロードすることはできません。また、アップロードされたパートは削除されます。

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 環境変数からアクセス認証情報を取得します。サンプルコードを実行する前に、OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# オブジェクトの完全なパスを指定します。完全なパスにバケット名を含めないでください。例: exampledir/exampleobject.txt。
key = 'exampledir/exampleobject.txt'
# アップロード ID を指定します。アップロード ID は、InitiateMultipartUpload 操作へのレスポンスから取得できます。
upload_id = 'yourUploadId'

# 指定されたアップロード ID を使用してマルチパートアップロードタスクをキャンセルします。アップロードされたパートは削除されます。
bucket.abort_multipart_upload(key, upload_id)

アップロードされたパートをリストする

次のサンプルコードは、アップロードされたパートをリストする方法の例を示しています。

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 環境変数からアクセス認証情報を取得します。サンプルコードを実行する前に、OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# オブジェクトの完全なパスを指定します。完全なパスにバケット名を含めないでください。例: exampledir/exampleobject.txt。
key = 'exampledir/exampleobject.txt'
# アップロード ID を指定します。アップロード ID は、InitiateMultipartUpload 操作へのレスポンスから取得できます。CompleteMultipartUpload 操作を呼び出してマルチパートアップロードタスクを完了する前に、アップロード ID を取得する必要があります。
upload_id = 'yourUploadId'

# 指定されたアップロード ID を使用するアップロードされたパートをリストします。
for part_info in oss2.PartIterator(bucket, key, upload_id):
    print('part_number:', part_info.part_number)
    print('etag:', part_info.etag)
    print('size:', part_info.size)

マルチパートアップロードタスクをリストする

特定のオブジェクトのマルチパートアップロードタスクをリストする

次のサンプルコードは、特定のオブジェクトのマルチパートアップロードタスクをリストする方法の例を示しています。

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 環境変数からアクセス認証情報を取得します。サンプルコードを実行する前に、OSS_ACCESS_KEY_ID および OSS_ACCESS_KEY_SECRET 環境変数が設定されていることを確認してください。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# オブジェクトの完全なパスを指定します。完全なパスにバケット名を含めないでください。例: exampledir/exampleobject.txt。
key = 'exampledir/exampleobject.txt'

# オブジェクトのすべてのマルチパートアップロードタスクをリストします。同じオブジェクトに対して init_multipart_upload メソッドが使用されるたびに、異なるアップロード ID が返されます。
# アップロード ID は、マルチパートアップロードタスクを一意に識別します。
for upload_info in oss2.ObjectUploadIterator(bucket, key):
    print('key:', upload_info.key)
    print('upload_id:', upload_info.upload_id)

バケット内のすべてのマルチパートアップロードタスクをリストする

次のサンプルコードは、バケット内のすべてのマルチパートアップロードタスクをリストする方法の例を示しています。

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。これらの認証情報を使用して OSS で操作を実行することは、リスクの高い操作です。RAM ユーザーを使用して API 操作を呼び出したり、日常的な O&M を実行することをお勧めします。RAM ユーザーを作成するには、RAM コンソールにログインします。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# バケット内のすべてのマルチパートアップロードタスクをリストします。
for upload_info in oss2.MultipartUploadIterator(bucket):
    print('key:', upload_info.key)
    print('upload_id:', upload_info.upload_id)

バケット内で名前が特定のプレフィックスを含むオブジェクトのマルチパートアップロードタスクをリストする

次のサンプルコードは、バケット内で名前が特定のプレフィックスを含むオブジェクトのマルチパートアップロードタスクをリストする方法の例を示しています。

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。これらの認証情報を使用して OSS で操作を実行することは、リスクの高い操作です。RAM ユーザーを使用して API 操作を呼び出したり、日常的な O&M を実行することをお勧めします。RAM ユーザーを作成するには、RAM コンソールにログインします。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# バケットが配置されているリージョンのエンドポイントを指定します。たとえば、バケットが中国 (杭州) リージョンにある場合は、エンドポイントを https://oss-cn-hangzhou.aliyuncs.com に設定します。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"

# エンドポイントにマップするリージョンの ID を指定します。例: cn-hangzhou。署名アルゴリズム V4 を使用する場合は、このパラメーターが必要です。
region = "cn-hangzhou"

# バケットの名前を指定します。
bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# バケット内で名前が test プレフィックスを含むオブジェクトのマルチパートアップロードタスクをリストします。
for upload_info in oss2.MultipartUploadIterator(bucket, prefix='test'):
    print('key:', upload_info.key)
    print('upload_id:', upload_info.upload_id)

よくある質問

パートを削除するにはどうすればよいですか?

マルチパートアップロードタスクが中断され、AbortMultipartUpload 操作が呼び出されない場合、タスクによってアップロードされたパートは指定されたバケットに保存されます。パートが不要になった場合は、次のいずれかの方法を使用してパートを削除し、不要なストレージコストを回避できます。

パートを手動で削除します。詳細については、「パートを削除する」をご参照ください。
ライフサイクルルールを設定してパートを削除します。詳細については、「ライフサイクルルールを設定して期限切れのパートを削除する」をご参照ください。

参考資料

マルチパートアップロードには 3 つの API 操作が含まれます。操作の詳細については、次のトピックを参照してください。
マルチパートアップロードタスクをキャンセルするために呼び出すことができる API 操作の詳細については、「AbortMultipartUpload」をご参照ください。
アップロードされたパートをリストするために呼び出すことができる API 操作の詳細については、「ListParts」をご参照ください。
進行中のすべてのマルチパートアップロードタスク (開始済みだが完了またはキャンセルされていないタスク) をリストするために呼び出すことができる API 操作の詳細については、「ListMultipartUploads」をご参照ください。