Compartir a través de

Carga de un blob en bloques con Python

  • Artículo

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

Configurar el entorno

Si no tiene un proyecto existente, en esta sección se muestra cómo configurar uno para que funcione con la biblioteca cliente de Azure Blob Storage para Python. Para más información, vea Introducción a Azure Blob Storage y Python.

Para trabajar con los ejemplos de código de este artículo, siga los pasos siguientes a fin de configurar el proyecto.

Instalar paquetes

Instale los siguientes paquetes mediante pip install:

pip install azure-storage-blob azure-identity

Adición de instrucciones import

Agregue las instrucciones siguientes import :

import io
import os
import uuid
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient, ContainerClient, BlobBlock, BlobClient, StandardBlobTier

Authorization

El mecanismo de autorización debe tener los permisos necesarios para actualizar un blob. Para la autorización con Microsoft Entra ID (recomendado) se necesita el rol integrado de RBAC de Azure de Colaborador de datos de Storage Blob o superior. Para más información, consulte la guía de autorización para Put Blob (API REST) y Put Block (API REST).

Creación de un objeto de cliente

Para conectar una aplicación a Blob Storage, cree una instancia de BlobServiceClient. En el ejemplo siguiente se muestra cómo crear un objeto de cliente mediante DefaultAzureCredential para la autorización:

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

También puede crear objetos de cliente para contenedores o blobs específicos ya sea directamente o desde el objeto BlobServiceClient. Para obtener más información sobre cómo crear y administrar objetos de cliente, consulte Crear y administrar objetos de cliente que interactúan con los recursos de datos.

Carga de datos en un blob en bloques

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

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:

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

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:

  1. Agregue las siguientes instrucciones import:

    import asyncio

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

  2. 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())

  3. 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.

Ejemplos de código

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:

Recursos de la biblioteca cliente

Consulte también

Contenido relacionado

  • Este artículo forma parte de la guía para desarrolladores de Blob Storage para Python. Para más información, consulte la lista completa de artículos de la guía para desarrolladores en Compilación de la aplicación de Python.