Carregar um blob de blocos com Python

Artigo
11/20/2023

Este artigo mostra como carregar um blob usando a Biblioteca de clientes do Armazenamento do Azure para Python. Você pode carregar dados em um blob de blocos de um caminho de arquivo, um fluxo, um objeto binário ou uma cadeia de caracteres de texto. Você também pode carregar blobs com marcas de índice.

Para saber mais sobre como carregar blobs usando APIs assíncronas, confira Carregar blobs de forma assíncrona.

Pré-requisitos

Este artigo pressupõe que você já tenha um projeto configurado para trabalhar com a biblioteca de clientes do Armazenamento de Blobs do Azure para Python. Para saber mais sobre como configurar seu projeto, incluindo a instalação de pacote, a adição de instruções de import e a criação de um objeto de cliente autorizado, consulte Introdução ao Armazenamento de Blobs do Azure e Python.
Para usar APIs assíncronas no seu código, confira os requisitos na seção Programação assíncrona.
O mecanismo de autorização deve ter permissões para executar uma operação de upload. Para saber mais, confira as diretrizes de autorização para as seguintes operações de API REST:
- Put Blob
- Put Block

Carregar dados em um blob de blocos

Para carregar um blob usando um fluxo ou um objeto binário, use o seguinte método:

upload_blob

Esse método cria um novo blob de uma fonte de dados com fragmentação automática, o que significa que a fonte de dados pode ser dividida em partes menores e carregada. Para realizar o carregamento, a biblioteca de clientes pode usar Put Blob ou uma série de chamadas Put Block seguidas de Put Block List. Esse comportamento depende do tamanho geral do objeto e de como as opções de transferência de dados são definidas.

Carregar um blob de blocos de um caminho de arquivo local

O seguinte exemplo carrega um arquivo em um blob de blocos usando um objeto BlobClient:

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)

Carregar um blob de blocos de um fluxo

O seguinte exemplo cria bytes aleatórios de dados e carrega um objeto BytesIO em um blob de blocos usando um objeto BlobClient:

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")

Carregar dados binários em um blob de blocos

O seguinte exemplo carrega dados binários em um blob de blocos usando um objeto BlobClient:

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")

Carregar um blob de blocos com marcas de índice

O seguinte exemplo carrega um blob de blocos com marcas de índice:

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)

Carregar um blob de blocos com opções de configuração

Você pode definir opções de configuração da biblioteca de clientes ao carregar um blob. Essas opções podem ser ajustadas para melhorar o desempenho, melhorar a confiabilidade e otimizar os custos. Os exemplos de código a seguir mostram como definir opções de configuração para um carregamento no nível do método e no nível do cliente ao instanciar BlobClient. Essas opções também podem ser configuradas para uma instância ContainerClient ou uma instância BlobServiceClient.

Especifique as opções de transferência de dados para carregamento

Você pode definir opções de configuração ao instanciar um cliente para otimizar o desempenho das operações de transferência de dados. Você pode passar os seguintes argumentos de palavra-chave ao construir um objeto cliente no Python:

max_block_size - O tamanho máximo da parte para carregar um blob de bloco em partes. O padrão é 4 MiB.
max_single_put_size - Se o tamanho do blob for menor ou igual a max_single_put_size, o blob será carregado com uma única solicitação Put Blob. Se o tamanho do blob for maior que max_single_put_size ou desconhecido, o blob será carregado em partes usando Put Block e será confirmado usando Put Block List. O padrão é 64 MiB.

Para obter mais informações sobre limites de tamanho de transferência para Armazenamento de Blobs, confira Escalar destinos para armazenamento de blob.

Para operações de carregamento, você também pode passar o argumento max_concurrency ao chamar upload_blob. Esse argumento define o número máximo de conexões paralelas a serem usadas quando o tamanho do blob exceder 64 MiB.

O exemplo de código a seguir mostra como especificar opções de transferência de dados ao criar um objeto BlobClient e como carregar dados usando esse objeto cliente. Os valores fornecidos neste exemplo não se destinam a ser uma recomendação. Para ajustar adequadamente esses valores, você precisa considerar as necessidades específicas do seu aplicativo.

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)

Para saber mais sobre como ajustar as opções de transferência de dados, confira Ajuste de desempenho para uploads e downloads com o Python.

Definir a camada de acesso de um blob durante o upload

Você pode definir a camada de acesso de um blob no carregamento passando o argumento de palavra-chave standard_blob_tier para upload_blob. O Armazenamento do Azure oferece diferentes camadas de acesso para que você possa armazenar seus dados de blob da maneira mais econômica com base em como eles estão sendo usados.

O exemplo de código a seguir mostra como definir a camada de acesso ao carregar um 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)

A configuração da camada de acesso somente é permitida para blobs de blocos. Você pode definir a camada de acesso para um blob de blocos como Hot, Cool, Cold ou Archive. Para definir a camada de acesso como Cold, você deve usar no mínimo a versão 12.15.0 da biblioteca de clientes.

Para saber mais sobre as camadas de acesso, confira Visão geral das camadas de acesso.

Faça upload de um blob de blocos preparando os blocos e confirmando

Você pode ter maior controle sobre como dividir os uploads em blocos organizando manualmente os blocos individuais de dados. Quando todos os blocos que formam um blob estiverem preparados, você poderá fazer commit deles no Armazenamento de Blobs.

Use o seguinte método para criar um novo bloco a ser confirmado como parte de um blob:

stage_block

Use o seguinte método para gravar um blob especificando a lista de IDs de bloco que compõem o blob:

commit_block_list

O exemplo a seguir lê dados de um arquivo e prepara blocos a serem confirmados como parte de um 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)

Carregar blobs de forma assíncrona

A biblioteca de clientes do Armazenamento de Blobs do Azure para Python dá suporte ao carregamento de blobs de forma assíncrona. Para saber mais sobre os requisitos de instalação do projeto, confira Programação assíncrona.

Siga estas etapas para carregar um blob usando APIs assíncronas:

Adicione as seguintes instruções de importação:

import asyncio

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

Adicione o código para executar o programa usando asyncio.run. Essa função executa a corrotina passada, main() no nosso exemplo, e gerencia o loop de eventos asyncio. As corrotinas são declaradas com a sintaxe async/await. Neste exemplo, a corrotina main() primeiro cria o BlobServiceClient de nível superior usando async with e, em seguida, chama o método que carrega o blob. Observe que somente o cliente de nível superior precisa usar async with, pois os outros clientes criados a partir dele compartilham o mesmo pool de conexões.
```
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())
```

Adicione o código para carregar o blob. O exemplo a seguir carrega um blob de um caminho de arquivo local usando um objeto ContainerClient. O código é igual ao exemplo síncrono, exceto que o método é declarado com a palavra-chave async e a palavra-chave await é usada ao chamar o método 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)

Com essa configuração básica em vigor, você pode implementar outros exemplos neste artigo como corrotinas usando sintaxe a async/await.

Recursos

Para saber mais sobre como carregar blobs usando a biblioteca de clientes do Armazenamento de Blobs do Azure para Python, consulte os recursos a seguir.

Operações da API REST

O SDK do Azure para Python contém bibliotecas que se baseiam na API REST do Azure, permitindo a interação com as operações da API REST por meio de paradigmas conhecidos do Python. Os métodos da biblioteca de clientes para carregar blobs usam as seguintes operações da API REST: