使用 Python 上傳區塊 Blob

發行項
03/22/2024

本文說明如何使用適用於 Python 的 Azure 儲存體用戶端程式庫來上傳 Blob。您可以從檔案路徑、資料流、二進位物件或文字字串將資料上傳至區塊 Blob。也可以使用索引標籤上傳 Blob。

若要了解如何使用非同步 API 上傳 Blob，請參閱以非同步方式上傳 Blob。

必要條件

本文假設您已有專案設定為使用適用於 Python 的 Azure Blob 儲存體用戶端程式庫。若要了解如何設定您的專案，包括安裝套件、新增 import 陳述式，以及建立授權的用戶端物件，請參閱開始使用 Azure Blob 儲存體和 Python。
若要在程式代碼中使用非同步 API，請參閱非同步程式設計一節中的需求。
授權機制必須具有執行上傳作業的權限。若要深入了解，請參閱下列 REST API 作業的授權指導：
- Put Blob \(英文\)
- 放置區塊

將資料上傳至區塊 Blob

若要使用資料流或二進位物件上傳 Blob，請使用下列方法：

upload_blob

這個方法會從具有自動區塊化的資料來源建立新的 Blob，這表示資料來源可能會分割成較小的區塊並上傳。若要執行上傳，用戶端程式庫可能使用放置區塊或一系列放置區域呼叫，後面接著放置區塊清單。此行為取決於 Blob 的總大小，以及資料傳輸選項的設定方式。

從本機檔案路徑上傳區塊 Blob

下列範例會使用 BlobClient 物件，將檔案上傳至區塊 Blob：

def upload_blob_file(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = container_client.upload_blob(name="sample-blob.txt", data=data, overwrite=True)

從資料流上傳區塊 Blob

下列範例會建立資料的隨機位元組，並使用 BlobClient 物件，將 BytesIO 物件上傳至區塊 Blob：

def upload_blob_stream(self, blob_service_client: BlobServiceClient, container_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    input_stream = io.BytesIO(os.urandom(15))
    blob_client.upload_blob(input_stream, blob_type="BlockBlob")

將二進位資料上傳至區塊 Blob

下列範例會使用 BlobClient 物件，將二進位資料上傳至區塊 Blob：

def upload_blob_data(self, blob_service_client: BlobServiceClient, container_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob="sample-blob.txt")
    data = b"Sample data for blob"

    # Upload the blob data - default blob type is BlockBlob
    blob_client.upload_blob(data, blob_type="BlockBlob")

使用索引標籤上傳區塊 Blob

下列範例會使用索引標籤上傳區塊 Blob：

def upload_blob_tags(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    sample_tags = {"Content": "image", "Date": "2022-01-01"}
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = container_client.upload_blob(name="sample-blob.txt", data=data, tags=sample_tags)

使用設定選項上傳區塊 Blob

您可以在上傳 Blob 時定義用戶端程式庫設定選項。這些選項也可進行調整，以改善效能、增強可靠性，並將成本最佳化。下列程式碼範例示範如何在具現化 BlobClient 時，在方法層級和用戶端層級定義上傳的設定選項。您也可以為 ContainerClient 執行個體或 BlobServiceClient 執行個體設定這些選項。

指定用於上傳的資料傳輸選項

您可以在具現化用戶端時設定組態選項，以最佳化資料傳輸作業的效能。在 Python 中建構用戶端物件時，您可以傳遞下列關鍵字引數：

max_block_size - 以區塊形式上傳區塊 Blob 的區塊大小上限。預設為 4 MiB。
max_single_put_size - 如果 Blob 大小小於或等於 max_single_put_size，Blob 會以單一 Put Blob 要求上傳。如果 Blob 大小大於 max_single_put_size 或未知，則 Blob 會使用 Put Block 以區塊形式上傳，並使用 Put Block List 進行認可。預設為 64 MiB。

如需 Blob 儲存體傳輸大小限制的詳細資訊，請參閱調整 Blob 儲存體的目標。

針對上傳作業，您也可以在呼叫 upload_blob 時傳遞 max_concurrency 引數。這個引數會定義要在 Blob 大小超出 64 MiB 時使用的平行連線數目上限。

下列程式碼範例說明如何在建立 BlobClient 物件時指定資料傳輸選項，以及如何使用該用戶端物件上傳資料。此範例中提供的值並非旨在作為建議。若要正確調整這些值，您必須考慮應用程式的特定需求。

def upload_blob_transfer_options(self, account_url: str, container_name: str, blob_name: str):
    # Create a BlobClient object with data transfer options for upload
    blob_client = BlobClient(
        account_url=account_url, 
        container_name=container_name, 
        blob_name=blob_name,
        credential=DefaultAzureCredential(),
        max_block_size=1024*1024*4, # 4 MiB
        max_single_put_size=1024*1024*8 # 8 MiB
    )
    
    with open(file=os.path.join(r'file_path', blob_name), mode="rb") as data:
        blob_client = blob_client.upload_blob(data=data, overwrite=True, max_concurrency=2)

若要深入了解微調資料傳輸選項，請參閱使用 Python 進行上傳和下載的效能微調。

設定 Blob 在上傳時的存取層

您可以將 standard_blob_tier 關鍵字引數傳遞至 upload_blob，設定 Blob 在上傳時的存取層。 Azure 儲存體提供不同的存取層 (英文)，可讓您根據 Blob 資料的使用方式，以最具成本效益的方式來儲存 Blob 資料。

下列程式碼範例顯示如何在上傳 Blob 時設定存取層：

def upload_blob_access_tier(self, blob_service_client: BlobServiceClient, container_name: str, blob_name: str):
    blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)
    
    #Upload blob to the cool tier
    with open(file=os.path.join(r'file_path', blob_name), mode="rb") as data:
        blob_client = blob_client.upload_blob(data=data, overwrite=True, standard_blob_tier=StandardBlobTier.COOL)

只有區塊 Blob 才能設定存取層。您可以將區塊 Blob 的存取層設定為 Hot、Cool、 Cold 或 Archive。若要將存取層設定為 Cold，您必須最少使用用戶端程式庫版本 12.15.0。

若要深入了解存取層，請參閱存取層概觀。

使用暫存區塊和認可上傳區塊 Blob

您可以手動暫存個別資料區塊，以進一步控制如何將上傳項目分割成區塊。當組成 Blob 的所有區塊都暫存時，您可以將這些區塊認可至 Blob 儲存體。

請使用下列方法來建立要認可為 Blob 一部分的新區塊：

stage_block

請使用下列方法，透過指定組成 Blob 的區塊識別碼清單來寫入 Blob：

commit_block_list

下列範例會從檔案讀取資料，並暫存要認可為 Blob 一部分的區塊：

def upload_blocks(self, blob_container_client: ContainerClient, local_file_path: str, block_size: int):
    file_name = os.path.basename(local_file_path)
    blob_client = blob_container_client.get_blob_client(file_name)

    with open(file=local_file_path, mode="rb") as file_stream:
        block_id_list = []

        while True:
            buffer = file_stream.read(block_size)
            if not buffer:
                break

            block_id = uuid.uuid4().hex
            block_id_list.append(BlobBlock(block_id=block_id))

            blob_client.stage_block(block_id=block_id, data=buffer, length=len(buffer))

        blob_client.commit_block_list(block_id_list)

以非同步方式上傳 Blob

適用於 Python 的 Azure Blob 儲存體用戶端程式庫支援以非同步方式上傳 Blob。若要深入了解專案設定需求，請參閱非同步程式設計。

請遵循下列步驟，使用非同步 API 上傳 Blob：

新增下列匯入陳述式：

import asyncio

from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient, BlobClient, ContainerClient

新增程式碼以使用 asyncio.run 執行程式。此函式會執行傳遞的協同程式 (在我們的範例中為 main())，並管理 asyncio 事件迴圈。協同程式會以 async/await 語法宣告。在此範例中，main() 協同程式會先使用 async with 建立最上層 BlobServiceClient，然後呼叫可上傳 Blob 的方法。請注意，只有最上層用戶端需要使用 async with，因為從中建立的其他用戶端會共用相同的連線集區。

async def main():
    sample = BlobSamples()

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

    async with BlobServiceClient(account_url, credential=credential) as blob_service_client:
        await sample.upload_blob_file(blob_service_client, "sample-container")

if __name__ == '__main__':
    asyncio.run(main())

新增程式碼以上傳 Blob。下列範例會使用 ContainerClient 物件，從本機檔案路徑上傳 Blob。該程式碼與同步範例相同，不同之處在於該方法是以 async 關鍵字宣告的，而 await 關鍵字是在呼叫 upload_blob 方法時使用的。

async def upload_blob_file(self, blob_service_client: BlobServiceClient, container_name: str):
    container_client = blob_service_client.get_container_client(container=container_name)
    with open(file=os.path.join('filepath', 'filename'), mode="rb") as data:
        blob_client = await container_client.upload_blob(name="sample-blob.txt", data=data, overwrite=True)

完成這個基本設定後，您可以使用 async/await 語法，將本文中的其他範例實作為協同程式。

資源

若要深入了解如何使用適用於 Python 的 Azure Blob 儲存體用戶端程式庫來上傳 Blob，請參閱下列資源。

REST API 操作

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

Share via