Python を使用して非同期スケジュールを設定して BLOB をコピーする

  • [アーティクル]

この記事では、Python 用の Azure Storage クライアント ライブラリを使用して非同期スケジュールを設定して BLOB をコピーする方法について説明します。 BLOB は、同じストレージ アカウント内のソースからでも、別のストレージ アカウント内のソースからでも、特定の URL に対する HTTP GET 要求を介して取得されたアクセス可能なオブジェクトからでもコピーできます。 保留中のコピー操作を中止することもできます。

この記事で説明するクライアント ライブラリ メソッドでは、Copy Blob REST API 操作が使用されます。また、このメソッドは、非同期スケジュールを設定してコピーを実行する場合に使用できます。 ストレージ アカウントにデータを移動し、ソース オブジェクトの URL がわかっているほとんどのコピー シナリオについては、「Python を使用してソース オブジェクト URL から BLOB をコピーする」を参照してください。

前提条件

環境を設定する

既存のプロジェクトがない場合、Python 用 Azure Blob Storage クライアント ライブラリを操作するためのプロジェクトをセットアップする方法についてこのセクションで説明します。 詳細については、「Azure Blob Storage と Python で作業を開始する」を参照してください。

この記事のコード例を使用するには、次の手順に従ってプロジェクトを設定します。

パッケージをインストールする

pip install を使用して次のパッケージをインストールしてください。

pip install azure-storage-blob azure-identity

import ステートメントを追加する

次の import ステートメントを追加します。

import datetime
from azure.identity import DefaultAzureCredential
from azure.storage.blob import (
    BlobServiceClient,
    BlobClient,
    BlobLeaseClient,
    BlobSasPermissions,
    generate_blob_sas
)

承認

認可メカニズムには、コピー操作を実行または保留中のコピーを中止するために必要な権限が必要です。 Microsoft Entra ID を使用した認可 (推奨) の場合、最小特権の Azure RBAC 組み込みロールは、いくつかの要因によって異なります。 詳しくは、「Copy Blob (REST API)」または「Abort Copy Blob (REST API)」の認可ガイダンスを参照してください。

クライアント オブジェクトの作成

アプリを Blob Storage に接続するには、 BlobServiceClientのインスタンスを作成します。 次の例では、認可のために DefaultAzureCredential を使用してクライアント オブジェクトを作成する方法を示します。

# TODO: Replace <storage-account-name> with your actual storage account name
account_url = "https://<storage-account-name>.blob.core.windows.net"
credential = DefaultAzureCredential()

# Create the BlobServiceClient object
blob_service_client = BlobServiceClient(account_url, credential=credential)

また、特定のコンテナーまたは BLOB 用のクライアント オブジェクトを、直接または BlobServiceClient オブジェクトから作成することもできます。 クライアント オブジェクトの作成と管理の詳細については、「データ リソースを操作するクライアント オブジェクトを作成および管理する」を参照してください。

非同期スケジュールを設定した BLOB のコピーについて

Copy Blob 操作は、非同期的に完了でき、ベスト エフォートで実行されます。つまり、即座に開始することも、指定の時間枠内に完了することも保証されていません。 コピー操作はバックグラウンドでスケジュールされ、サーバーに使用可能なリソースがある場合に実行されます。 コピーが同じストレージ アカウント内で行われる場合、操作は同期的に完了できます。

Copy Blob 操作では、次のいずれかのアクションを実行できます。

  • コピー元 BLOB を別の名前でコピー先 BLOB にコピーします。 コピー先 BLOB は、同じ BLOB の種類 (ブロック、アペンド、またはページ) の既存の BLOB でも、コピー操作によって作成される新しい BLOB でもかまいません。
  • ソース BLOB を同じ名前でコピー先 BLOB にコピーして、コピー先 BLOB を置き換えます。 この種のコピー操作では、コミットされていないブロックはすべて削除され、コピー先 BLOB のメタデータが上書きされます。
  • Azure File Service 内のコピー元ファイルをコピー先 BLOB にコピーします。 コピー先 BLOB は、既存のブロック BLOB でも、コピー操作によって作成される新しいブロック BLOB でもかまいません。 ファイルからページ BLOB またはアペンド BLOB へのコピーはサポートされていません。
  • スナップショットをベース BLOB にコピーします。 スナップショットをベース BLOB に昇格することにより、BLOB を以前のバージョンに復元できます。
  • スナップショットを別の名前でコピー先 BLOB にコピーします。 結果として得られるコピー先 BLOB は書き込み可能な BLOB であり、スナップショットではありません。

コピー操作のソース BLOB にすることができるのは、ブロック BLOB、アペンド BLOB、ページ BLOB、BLOB のスナップショット、または BLOB のバージョンのいずれかの種類になります。 コピー操作では、常にソース BLOB またはファイル全体がコピーされます。 バイト範囲またはブロックのセットのコピーはサポートされていません。

コピー先 BLOB が既に存在する場合は、ソース BLOB と同じ種類の BLOB である必要があり、また既存のコピー先 BLOB は上書きされます。 コピー操作の進行中はコピー先 BLOB を変更できません。また、コピー先 BLOB には未処理のコピー操作が 1 つだけ存在することができます。

プロパティ、インデックス タグ、メタデータ、課金に関する情報など、Copy Blob 操作の詳細については、BLOB リマークのコピーに関するページを参照してください。

非同期スケジュールを設定して BLOB をコピーする

このセクションでは、非同期スケジュールを設定してコピー操作を実行するために、Python 用の Azure Storage クライアント ライブラリによって提供されるメソッドの概要について説明します。

次のメソッドは、Copy Blob REST API 操作をラップし、ソース BLOB からのデータの非同期コピーを開始します。

start_copy_from_url は、copy_status と copy_id を含む辞書を返します。 copy_status プロパティは、コピーが同期的に完了した場合は success、コピーが非同期的に開始された場合は pending になります。

Azure 内のソースから BLOB をコピーする

同じストレージ アカウント内の BLOB をコピーする場合、操作は同期的に完了できます。 ソース BLOB へのアクセスは、Microsoft Entra ID、Shared Access Signature (SAS)、またはアカウント キーを使用して承認できます。 変更可能な同期コピー操作については、「Python を使用してソース オブジェクト URL から BLOB をコピーする」を参照してください。

コピー ソースが別のストレージ アカウント内の BLOB である場合、操作は非同期的に完了できます。 ソース BLOB は、パブリックであるか、SAS トークンを介して認可されている必要があります。 SAS トークンには Read ('r') アクセス許可を含める必要があります。 SAS トークンの詳細については、「Shared Access Signatures によるアクセスの委任」を参照してください。

次の例は、非同期スケジュールを使用して別のストレージ アカウントからソース BLOB をコピーするシナリオを示しています。 この例では、追加されたユーザー委任 SAS トークンを使用してソース BLOB URL を作成します。 この例では、クライアント ライブラリを使用して SAS トークンを生成する方法を示していますが、独自のトークンを指定することもできます。 この例では、コピー操作中にソース BLOB をリースして、別のクライアントから BLOB が変更されないようにする方法も示します。 Copy Blob 操作では、コピー操作の開始時にソース BLOB の ETag 値が保存されます。 コピー操作が完了する前に ETag 値が変更されると、操作は失敗します。

def copy_from_source_in_azure_async(self, source_blob: BlobClient, destination_blob: BlobClient, blob_service_client: BlobServiceClient):
    # Lease the source blob during copy to prevent other clients from modifying it
    lease = BlobLeaseClient(client=source_blob)

    sas_token = self.generate_user_delegation_sas(blob_service_client=blob_service_client, source_blob=source_blob)
    source_blob_sas_url = source_blob.url + "?" + sas_token

    # Create an infinite lease by passing -1 as the lease duration
    lease.acquire(lease_duration=-1)

    # Start the copy operation - specify False for the requires_sync parameter
    copy_operation = dict()
    copy_operation = destination_blob.start_copy_from_url(source_url=source_blob_sas_url, requires_sync=False)
    
    # If start_copy_from_url returns copy_status of 'pending', the operation has been started asynchronously
    # You can optionally add logic here to wait for the copy operation to complete

    # Release the lease on the source blob
    lease.break_lease()

def generate_user_delegation_sas(self, blob_service_client: BlobServiceClient, source_blob: BlobClient):
    # Get a user delegation key
    delegation_key_start_time = datetime.datetime.now(datetime.timezone.utc)
    delegation_key_expiry_time = delegation_key_start_time + datetime.timedelta(hours=1)
    key = blob_service_client.get_user_delegation_key(
        key_start_time=delegation_key_start_time,
        key_expiry_time=delegation_key_expiry_time
    )

    # Create a SAS token that's valid for one hour, as an example
    sas_token = generate_blob_sas(
        account_name=blob_service_client.account_name,
        container_name=source_blob.container_name,
        blob_name=source_blob.blob_name,
        account_key=None,
        user_delegation_key=key,
        permission=BlobSasPermissions(read=True),
        expiry=datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1),
        start=datetime.datetime.now(datetime.timezone.utc)
    )

    return sas_token

Note

ユーザー委任 SAS トークンでは、アカウント キーではなく Microsoft Entra の資格情報で署名されているため、セキュリティが強化されます。 ユーザー委任 SAS トークンを作成するには、Microsoft Entra セキュリティ プリンシパルに適切なアクセス許可が必要です。 認可要件については、「ユーザー委任キーを取得する」を参照してください。

Azure 外部のソースから BLOB をコピーする

Azure 外部にあるアクセス可能なオブジェクトを含め、特定の URL に対する HTTP GET 要求を介して取得できる任意のソース オブジェクトにコピー操作を実行できます。 次の例は、アクセス可能なソース オブジェクト URL から BLOB をコピーするシナリオを示しています。

def copy_from_external_source_async(self, source_url: str, destination_blob: BlobClient):
    # Start the copy operation - specify False for the requires_sync parameter
    copy_operation = dict()
    copy_operation = destination_blob.start_copy_from_url(source_url=source_url, requires_sync=False)
    
    # If start_copy_from_url returns copy_status of 'pending', the operation has been started asynchronously
    # You can optionally add logic here to wait for the copy operation to complete

コピー操作の状態を確認する

非同期の Copy Blob 操作の状態を確認するには、get_blob_properties メソッドをポーリングして、コピー状態を確認します。

次のコード例は、保留中のコピー操作の状態を確認する方法を示しています。

def check_copy_status(self, destination_blob: BlobClient):
    # Get the copy status from the destination blob properties
    copy_status = destination_blob.get_blob_properties().copy.status

    return copy_status

コピー操作を中止する

保留中の Copy Blob 操作を中止すると、コピー先 BLOB の長さは 0 になります。 ただし、コピー先 BLOB のメタデータは、ソース BLOB からコピーされた値、またはコピー操作中に明示的に設定された値に変わります。 コピー前のメタデータを元のまま維持するには、いずれかのコピー方法を呼び出す前に、コピー先 BLOB のスナップショットを作成します。

保留中のコピー操作を中止するには、次の操作を呼び出します。

このメソッドは、保留中の Copy Blob 操作を取り消す Abort Copy Blob REST API 操作をラップします。 次のコード例は、保留中の Copy Blob 操作を中止する方法を示しています。

def abort_copy(self, destination_blob: BlobClient):
    # Get the copy operation details from the destination blob properties
    copy_status = destination_blob.get_blob_properties().copy.status
    copy_id = destination_blob.get_blob_properties().copy.id

    # Check the copy status and abort if pending
    if copy_status == 'pending':
        destination_blob.abort_copy(copy_id)
        print(f"Copy operation {copy_id} has been aborted")

リソース

Python 用 Azure Blob Storage クライアント ライブラリを使用して非同期スケジュールを設定して BLOB をコピーする方法の詳細については、次のリソースを参照してください。

コード サンプル

REST API の操作

Azure SDK for Python には Azure REST API に基づき構築されたライブラリが含まれるため、使い慣れた Python パラダイムを通じて REST API 操作を実施できます。 このアーティクルで説明するクライアント ライブラリ メソッドでは、次の REST API 操作を使用します:

クライアント ライブラリのリソース

関連するコンテンツ

  • この記事は、Python の Blob Storage 開発者ガイドの一部です。 詳細については、「Python アプリの構築」にある開発者ガイドの記事の完全な一覧を参照してください。