OSS SDK for Python V2 を使用してファイルをコピーする方法 - Object Storage Service

このトピックでは、OSS SDK for Python V2 の Copier モジュールを使用して、大きなファイルを転送するためにファイルをコピーする方法について説明します。

注意事項

このトピックのサンプルコードでは、中国 (杭州) リージョンのリージョン ID cn-hangzhou を使用します。デフォルトではパブリックエンドポイントが使用されます。同じリージョン内の他の Alibaba Cloud サービスから OSS にアクセスする場合は、内部エンドポイントを使用してください。OSS リージョンとエンドポイントのマッピングの詳細については、「リージョンとエンドポイント」をご参照ください。
オブジェクトをコピーするには、ソースオブジェクトに対する読み取り権限と、宛先バケットに対する読み取りおよび書き込み権限が必要です。
クロスリージョンコピーはサポートされていません。たとえば、中国 (杭州) リージョンのバケットから中国 (青島) リージョンのバケットにオブジェクトをコピーすることはできません。
オブジェクトをコピーするときは、ソースバケットと宛先バケットにリテンションポリシーが設定されていないことを確認してください。そうしないと、The object you specified is immutable. エラーが返されます。

メソッド定義

コピーマネージャーの概要

あるバケットから別のバケットにオブジェクトをコピーしたり、オブジェクトのプロパティを変更したりするには、コピー操作またはマルチパートコピー操作を使用できます。各操作は、異なるシナリオに適しています。

コピー操作 (CopyObject) は、5 GiB 未満のオブジェクトのコピーにのみ適しています。
マルチパートコピー操作 (UploadPartCopy) は、5 GiB を超えるオブジェクトのコピーをサポートします。ただし、この操作は (x-oss-metadata-directive) または (x-oss-tagging-directive) パラメーターをサポートしていません。この操作を使用する場合は、コピーするメタデータとタグを指定する必要があります。

コピーマネージャー Copier は、OSS SDK for Python V2 の新機能で、汎用的なコピーインターフェイスを提供します。このインターフェイスは、基盤となる実装の詳細を抽象化し、リクエストパラメーターに基づいて適切なコピー操作を自動的に選択します。次のコードは、Copier の一般的なメソッドを示しています。

class CopyError(exceptions.BaseError):
  ...

def copier(self, **kwargs) -> Copier:
  ...

def copy(self, request: models.CopyObjectRequest, **kwargs: Any) -> CopyResult:
  ...

リクエストパラメーター

パラメーター	タイプ	説明
request	CopyObjectRequest	操作のリクエストパラメータ。詳細については、「CopyObjectRequest」をご参照ください。
**kwargs	Any	(オプション) 辞書型の任意のパラメーター。

次の表に、CopyObjectRequest の一般的なパラメーターを示します。

パラメーター	タイプ	説明
bucket	str	デスティネーションバケットの名前。
key	str	デスティネーションオブジェクトの名前。
source_bucket	str	ソースバケットの名前。
source_key	str	ソースオブジェクトの名前。
forbid_overwrite	str	CopyObject 操作中に同じ名前の宛先オブジェクトを上書きするかどうかを指定します。
tagging	str	オブジェクトのタグ。一度に複数のタグを指定できます。例: TagA=A&TagB=B。
tagging_directive	str	宛先オブジェクトのタグを設定する方法を指定します。有効な値: Copy (デフォルト): ソースオブジェクトのタグを宛先オブジェクトにコピーします。 Replace: ソースオブジェクトのタグを無視し、リクエストで指定されたタグを宛先オブジェクトに使用します。

client.copier を使用してコピーマネージャーインスタンスを初期化するときに構成オプションを指定することで、オブジェクトのコピー動作をカスタマイズできます。また、各コピー呼び出しに構成オプションを指定して、特定のオブジェクトの動作をカスタマイズすることもできます。

Copier の構成パラメーターを設定する

copier = client.copier(
    part_size=100 * 1024 * 1024, # パートサイズを設定
)

各コピーリクエストの構成パラメーターを設定する

result = copier.copy(oss.CopyObjectRequest(
        bucket="example_bucket", # コピー先バケット名
        key="example_key", # コピー先オブジェクト名
        source_bucket="example_source_bucket", # コピー元バケット名
        source_key="example_source_key", # コピー元オブジェクト名
    ),
    part_size=100 * 1024 * 1024, # パートサイズを設定
)

次の表に、一般的な構成オプションを示します。

パラメーター	タイプ	説明
part_size	int	パートサイズ。デフォルト値は 64 MiB です。
parallel_num	int	同時コピーのタスク数。デフォルト値: 3。このパラメーターは、グローバルな同時実行数制限ではなく、単一の呼び出しの同時実行数制限を指定します。
multipart_copy_threshold	int64	マルチパートコピーのしきい値。デフォルト値は 200 MiB です。
leave_parts_on_error	bool	コピーに失敗した場合にコピーされたパートを保持するかどうかを指定します。デフォルトでは、コピーされたパートは保持されません。
disable_shallow_copy	bool	シャローコピーを無効にするかどうかを指定します。デフォルトでは、シャローコピーが使用されます。

サンプルコード

次のサンプルコードは、ソースバケットから宛先バケットにオブジェクトをコピーする方法を示しています。

import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="copier 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)

# コマンドラインパラメーター source_key (必須) を追加します。これはソースオブジェクトの名前を指定します。
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# コマンドラインパラメーター source_bucket (必須) を追加します。これはソースバケットの名前を指定します。
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


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

    # 環境変数から認証情報を読み込みます。
    # EnvironmentVariableCredentialsProvider を使用して、環境変数から AccessKey ID と AccessKey Secret を読み取ります。
    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)

    # Copier インスタンスを作成し、オブジェクトのコピー操作を実行します。
    copier = client.copier()

    # オブジェクトのコピー操作を実行します。
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 宛先バケットの名前。
            key=args.key,                # 宛先オブジェクトの名前。
            source_bucket=args.source_bucket,  # ソースバケットの名前。
            source_key=args.source_key   # ソースオブジェクトの名前。
        )
    )

    # コピー結果を出力します。
    # vars(result) を使用して結果オブジェクトを辞書形式に変換し、結果を出力します。
    print(vars(result))


if __name__ == "__main__":
    main()

シナリオ

コピーマネージャーを使用してパートサイズと同時実行数を設定する

次のサンプルコードは、コピーマネージャーのパラメーターを構成して、パートサイズと同時実行数を設定する方法を示しています。

import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="copier 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)

# コマンドラインパラメーター source_key (必須) を追加します。これはソースオブジェクトの名前を指定します。
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# コマンドラインパラメーター source_bucket (必須) を追加します。これはソースバケットの名前を指定します。
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


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

    # 環境変数から認証情報を読み込みます。
    # EnvironmentVariableCredentialsProvider を使用して、環境変数から AccessKey ID と AccessKey Secret を読み取ります。
    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)

    # Copier インスタンスを作成し、オブジェクトのコピー操作を実行します。
    copier = client.copier()

    # オブジェクトのコピー操作を実行します。
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 宛先バケットの名前。
            key=args.key,                # 宛先オブジェクトの名前。
            source_bucket=args.source_bucket,  # ソースバケットの名前。
            source_key=args.source_key   # ソースオブジェクトの名前。
        ),
        part_size= 1 * 1024 * 1024,          # パートサイズ (バイト単位)。この例では、パートサイズは 1 MiB に設定されています。
        parallel_num=5,                 # 同時タスクの数。このパラメーターは、同時にコピーできるパートの数を指定します。
        leave_parts_on_error=True,      # コピーに失敗した場合、コピーされたパートを保持します。
    )

    # コピー結果を出力します。
    # vars(result) を使用して結果オブジェクトを辞書形式に変換し、結果を出力します。
    print(vars(result))


if __name__ == "__main__":
    main()

コピーマネージャーを使用して進捗横棒グラフを表示する

次のサンプルコードは、進捗横棒グラフを使用してコピーの進捗状況を表示する方法を示しています。

import argparse
import alibabacloud_oss_v2 as oss

# コマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="copier 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)

# コマンドラインパラメーター source_key (必須) を追加します。これはソースオブジェクトの名前を指定します。
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# コマンドラインパラメーター source_bucket (必須) を追加します。これはソースバケットの名前を指定します。
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


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

    # 環境変数から認証情報を読み込みます。
    # EnvironmentVariableCredentialsProvider を使用して、環境変数から AccessKey ID と AccessKey Secret を読み取ります。
    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 が現在の行を上書きできるようにします。
        print(f'\rCopy progress: {rate}% ', end='')

    # Copier インスタンスを作成し、オブジェクトのコピー操作を実行します。
    copier = client.copier()

    # オブジェクトのコピー操作を実行します。
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 宛先バケットの名前。
            key=args.key,                # 宛先オブジェクトの名前。
            source_bucket=args.source_bucket,  # ソースバケットの名前。
            source_key=args.source_key,   # ソースオブジェクトの名前。
            progress_fn=_progress_fn,   # 進捗コールバック関数を設定します。
        )
    )

    # コピー結果を出力します。
    # vars(result) を使用して結果オブジェクトを辞書形式に変換し、結果を出力します。
    print(vars(result))


if __name__ == "__main__":
    main()

Object Storage Service:ファイルコピーマネージャー (Python SDK V2)