使用 Python 複製具有非同步排程的 Blob

• .NET
• Java
• JavaScript
• Python
• Go

本文說明如何使用適用於 Python 的 Azure 儲存體用戶端程式庫，以非同步排程來複製 Blob。 您可以從相同儲存體帳戶內的來源、從不同儲存體帳戶中的來源，或從指定 URL 上透過 HTTP GET 要求擷取的任何可存取物件，來複製 Blob。 您也可以中止擱置中的複製作業。

本文涵蓋的用戶端程式庫方法會使用複製 Blob REST API 作業，而且當您想要以非同步排程執行複製時，可以加以使用。 針對您想要將資料移至儲存體帳戶且具有來源物件 URL 的大多數複製案例，請參閱使用 Python 從來源物件 URL 複製 Blob。

必要條件

• Azure 訂用帳戶 - 建立免費帳戶
• Azure 儲存體帳戶 - 建立儲存體帳戶
• Python 3.8+

設定您的環境

如果沒有現有的專案，本章節會說明如何設定專案以使用適用於 Python 的 Azure Blob 儲存體用戶端程式庫。 如需詳細資訊，請參閱開始使用 Azure Blob 儲存體和 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 內建角色會因數種因素而異。 若要深入了解，請參閱複製 Blob (REST API) 或中止複製 Blob (REST API) 的授權指導。

建立用戶端物件

若要將應用程式連線至 Blob 儲存體，請建立 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)

您也可以直接或從 BlobServiceClient 物件建立特定容器或 Blob 的用戶端物件。 若要深入了解如何建立及管理用戶端物件，請參閱建立和管理與資料資源互動的用戶端端物件 (部分機器翻譯)。

關於使用非同步排程複製 Blob

Copy Blob 作業可以非同步完成，而且會盡最大努力執行，這表示作業不保證會立即開始或在指定的時間範圍內完成。 複製作業會排程在背景中，並在伺服器具有可用的資源時執行。 如果複製發生在相同的儲存體帳戶內，作業可以同步完成。

Copy Blob 作業可以執行下列任何動作：

• 將來源 Blob 複製到具有不同名稱的目的地 Blob。 目的地 Blob 可以是相同 Blob 類型 (區塊、附加或分頁) 的現有 Blob，也可以是複製作業所建立的新 Blob。
• 將來源 Blob 複製到名稱相同的目的地 Blob，藉此取代目的地 Blob。 此類型的複製作業會移除任何未認可的區塊，並覆寫目的地 Blob 的中繼資料。
• 將 Azure 檔案服務中的來源檔案複製到目的地 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 只能有一個未完成的複製作業。

若要深入了解 Copy Blob 作業的相關資訊，包括屬性、索引標籤、中繼資料和計費，請參閱複製 Blob 備註。

使用非同步排程來複製 Blob

本節概述「適用於 Python 的 Azure 儲存體用戶端程式庫」所提供的方法，以使用非同步排程來執行複製作業。

下列方法會包裝複製 Blob REST API 作業，然後從來源 Blob 開始非同步複製資料：

• start_copy_from_url

start_copy_from_url 會傳回包含 copy_status 和 copy_id 的字典。 如果複製以同步方式完成，copy_status 屬性是 success，如果複製以非同步方式啟動，則該屬性是 pending。

從 Azure 中的來源複製 Blob

如果您要在相同的儲存體帳戶內複製 Blob，作業可以同步完成。 您可以透過 Microsoft Entra ID、共用存取簽章 (SAS) 或帳戶金鑰來授權來源 Blob 的存取權。 如需另一種同步複製作業，請參閱使用 Python 從來源物件 URL 複製 Blob。

如果複製來源是不同儲存體帳戶中的 Blob，作業可以非同步完成。 來源 Blob 必須透過 SAS 權杖公開或授權。 SAS 權杖需要包含讀取 ('r') 權限。 若要深入了解 SAS 權杖，請參閱使用共用存取簽章委派存取權。

下列範例示範使用非同步排程從不同儲存體帳戶複製來源 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

注意

使用者委派 SAS 權杖可提供更高的安全性，因為它們是以 Microsoft Entra 認證進行簽署，而不是帳戶金鑰。 若要建立使用者委派 SAS 權杖，Microsoft Entra 安全性主體需有適當的權限。 有關授權需求，請參閱取得使用者委派金鑰。

從 Azure 外部的來源複製 Blob

您可以在透過 HTTP GET 要求於指定 URL 擷取的任何來源物件上，執行複製作業，包括 Azure 外部的可存取物件。 下列範例示範從可存取的來源物件 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 的長度為零。 不過，目的地 Blob 的中繼資料會包含新值，這可能是從來源 Blob 複製而來的值，或是在複製作業期間明確設定的值。 若要保留複製前的原始中繼資料，請對目的地 Blob 進行快照集，再呼叫其中一個複製方法。

若要中止擱置中的複製作業，請呼叫下列作業：

• abort_copy

此方法會包裝中止複製 Blob REST API 作業，這會取消擱置中的 Copy Blob 作業。 下列程式碼範例說明如何中止擱置中的 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 儲存體用戶端程式庫，以透過非同步排程複製 Blob，請參閱下列資源。

程式碼範例

• 檢視本文中的程式碼範例 (GitHub) (英文)

REST API 操作

Azure SDK for Python 包含建置在 Azure REST API 之上的程式庫，可讓您透過熟悉的 Python 範例與 REST API 作業進行互動。 本文涵蓋的用戶端程式庫方法會使用下列 REST API 作業：

• 複製 Blob (REST API)
• 中止複製 Blob (REST API)

用戶端程式庫資源

• 用戶端程式庫參考文件
• 用戶端程式庫原始程式碼
• 套件 (Pypi) (英文)

相關內容

• 本文是適用於 Python 的 Blob 儲存體開發人員指南的一部分。 若要深入了解，請參閱 建置 Python 應用程式 中的開發人員指南文章完整清單。