Carga de un blob en bloques con Python

Artículo
11/17/2023

En este artículo se muestra cómo cargar un blob con la biblioteca cliente de Azure Storage para Python. Puede cargar datos en un blob en bloques desde una ruta de acceso de archivo, una secuencia, un objeto binario o una cadena de texto. También puede cargar blobs con etiquetas de índice.

Para obtener información sobre cómo cargar blobs mediante API asincrónicas, consulte Carga de blobs de forma asincrónica.

Requisitos previos

En este artículo se supone que ya tiene un proyecto configurado para trabajar con la biblioteca cliente de Azure Blob Storage para Python. Para obtener información sobre cómo configurar el proyecto, incluida la instalación de paquetes, la adición de directivas de import y la autorización de un objeto de cliente, consulte Introducción a Azure Storage y Java.
Para usar API asincrónicas en el código, consulte los requisitos de la sección Programación asincrónica.
El mecanismo de autorización debe tener permisos para realizar una operación de carga. Para obtener más información, consulte la guía de autorización para las siguientes operaciones de la API de REST:
- Put Blob
- Put Block

Carga de datos en un blob en bloques

Para cargar un blob mediante una secuencia o un objeto binario, use el método siguiente:

upload_blob

Este método crea un nuevo blob a partir de un origen de datos con fragmentación automática, lo que significa que el origen de datos se puede dividir en fragmentos más pequeños y cargarse. Para realizar la carga, la biblioteca cliente puede usar Put Blob o una serie de llamadas Put Block seguidas de Put Block List. Este comportamiento depende del tamaño total del objeto y de cómo se establezcan las opciones de transferencia de datos.

Carga de un blob en bloques desde una ruta de acceso de archivo local

En el ejemplo siguiente se carga un archivo a un blob en bloques mediante un 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)

Carga de un blob en bloques desde una secuencia

En el ejemplo siguiente se crean bytes aleatorios de datos y se carga un objeto BytesIO a un blob en bloques mediante un 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")

Cargar datos binarios en un blob en bloques

En el ejemplo siguiente se cargan datos binarios a un blob en bloques mediante un 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")

Carga de un blob en bloques con etiquetas de índice

En el ejemplo siguiente se carga un blob en bloques con etiquetas 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)

Carga de un blob en bloques con opciones de configuración

Puede definir las opciones de configuración de la biblioteca cliente al cargar un blob. Estas opciones se pueden ajustar para mejorar el rendimiento, mejorar la confiabilidad y optimizar los costes. En los ejemplos de código siguientes, se muestra cómo definir opciones de configuración para una carga tanto en el nivel de método como en el nivel de cliente al crear instancias de BlobClient. Estas opciones también se pueden configurar para una instancia ContainerClient o una instancia de BlobServiceClient.

Especificación de opciones de transferencia de datos para la carga

Puede establecer opciones de configuración al crear instancias de un cliente para optimizar el rendimiento de las operaciones de transferencia de datos. Puede pasar los argumentos de palabra clave siguientes al construir un objeto de cliente en Python:

max_block_size: tamaño máximo del fragmento para cargar un blob en bloques en fragmentos. El valor predeterminado es 4 MiB.
max_single_put_size: si el tamaño del blob es menor o igual que max_single_put_size, el blob se carga con una sola solicitud Put Blob. Si el tamaño del blob es superior a max_single_put_size o desconocido, el blob se carga en fragmentos usando Put Block y se confirma usando Put Block List. El valor predeterminado es 64 MiB.

Para más información sobre los límites de tamaño de transferencia para Blob Storage, consulte Objetivos de escalabilidad de Blob Storage.

Para las operaciones de carga, también puede pasar el argumento max_concurrency al llamar a upload_blob. Este argumento define el número máximo de conexiones paralelas que se usarán cuando el tamaño del blob supere los 64 MiB.

En el ejemplo de código siguiente, se muestra cómo especificar opciones de transferencia de datos al crear un objeto BlobClient y cómo cargar datos mediante ese objeto de cliente. Los valores proporcionados en este ejemplo no están diseñados para ser una recomendación. Para ajustar correctamente estos valores, debe tener en cuenta las necesidades específicas de la aplicación.

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 obtener más información sobre cómo ajustar las opciones de transferencia de datos, consulte Ajuste del rendimiento para cargas y descargas con Python.

Establecimiento del nivel de acceso de un blob durante la carga

Puede establecer el nivel de acceso de un blob al cargar pasando el argumento de palabra standard_blob_tier clave a upload_blob. Azure Storage ofrece diferentes niveles de acceso para que pueda almacenar los datos de blobs de la manera más rentable en función de cómo se usen.

En el ejemplo de código siguiente se muestra cómo establecer el nivel de acceso al cargar un 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)

Solo se permite establecer el nivel de acceso en los blobs en bloques. Puede establecer el nivel de acceso de un blob en bloques en Hot, Cool, Cold o Archive. Para establecer el nivel de acceso en Cold, debe usar la versión mínima de la biblioteca cliente 12.15.0.

Si desea obtener más información sobre los niveles de acceso, consulte Información general sobre los niveles de acceso.

Cargar un blob en bloques mediante la preparación de bloques y la confirmación

Puede tener un mayor control sobre cómo dividir las cargas en bloques si pone en escena manualmente bloques individuales de datos. Cuando todos los bloques que son un blob se han almacenado de manera provisional, puede confirmarlos en Blob Storage.

Use el método siguiente para crear un nuevo bloque que se confirmará como parte de un blob:

stage_block

Use el método siguiente para escribir un blob especificando la lista de identificadores de bloque que componen el blob:

commit_block_list

En el ejemplo siguiente se leen los datos de un archivo y los bloques de fases que se confirmarán como parte de un 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)

Carga de blobs de forma asincrónica

La biblioteca cliente de Azure Blob Storage para Python admite la carga de blobs de forma asincrónica. Para obtener más información sobre los requisitos de configuración del proyecto, consulte Programación asincrónica.

Siga estos pasos para cargar un blob mediante API asincrónicas:

Agregue las siguientes instrucciones import:

import asyncio

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

Agregue código para ejecutar el programa mediante asyncio.run. Esta función ejecuta la corrutina pasada, main() en nuestro ejemplo, y administra el bucle de eventos asyncio. Las corrutinas se declaran con la sintaxis async/await. En este ejemplo, la corrutina main() crea primero el nivel superior BlobServiceClient mediante async with y, a continuación, llama al método que carga el blob. Tenga en cuenta que solo el cliente de nivel superior debe usar async with, ya que otros clientes creados a partir de él comparten el mismo grupo de conexiones.
```
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())
```

Agregue código para cargar el blob. En el siguiente ejemplo se carga un blob desde una ruta de acceso de archivo local mediante un objeto ContainerClient. El código es el mismo que el ejemplo sincrónico, salvo que el método se declara con la palabra clave async y la palabra clave await se usa al llamar al 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)

Con esta configuración básica, puede implementar otros ejemplos de este artículo como corrutinas mediante la sintaxis async/await.

Recursos

Para obtener más información sobre cómo cargar blobs con la biblioteca cliente de Azure Blob Storage para Python, consulte los recursos siguientes.

Operaciones de API REST

El SDK de Azure para Python contiene bibliotecas que se crean a partir de la API REST de Azure, lo que le permite interactuar con las operaciones de API REST a través de paradigmas conocidos de Python. Los métodos de la biblioteca cliente para cargar blobs usan las siguientes operaciones de API REST: