SDK de Azure Cosmos DB para Python

Biblioteca de cliente para la API NoSQL de Azure Cosmos DB: base de datos multimodelo distribuida a nivel mundial.

Instalación

pip install azure-cosmos azure-identity

Variables de entorno

COSMOS_ENDPOINT=https://.documents.azure.com:443/  # Obligatorio para todos los métodos de autenticación
COSMOS_DATABASE=mydb  # Obligatorio para todos los métodos de autenticación
COSMOS_CONTAINER=mycontainer  # Requerido para todos los métodos de autenticación
AZURE_TOKEN_CREDENTIALS=prod # Requerido solo si se utiliza DefaultAzureCredential en producción

Autenticación y ciclo de vida

🔑 Hay dos reglas que se aplican a todos los ejemplos de código que aparecen a continuación:

1. Da prioridad a DefaultAzureCredential. Funciona tanto a nivel local (Azure CLI / VS Code / Developer CLI) como en Azure (identidad gestionada, identidad de carga de trabajo) sin necesidad de modificar el código. Evita las cadenas de conexión y las claves de cuenta o API, ya que eluden la auditoría y la rotación de Entra.
   • Desarrollo local: DefaultAzureCredential funciona tal cual.
   • Producción: establece AZURE_TOKEN_CREDENTIALS=prod (o AZURE_TOKEN_CREDENTIALS=) para limitar la cadena de credenciales a aquellas seguras para producción.
2. Envuelve cada cliente en un gestor de contexto para que los transportes HTTP, los sockets y las cachés de tokens se liberen de forma determinista:
   • Sincrónico: con (...) como cliente:
   • Asíncrono: async con (...) como cliente: y async con DefaultAzureCredential() como credencial: (de azure.identity.aio)

Los fragmentos de código pueden abreviar esta configuración, pero el código de producción siempre debe seguir ambas reglas.

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.cosmos import CosmosClient

# Desarrollo local: DefaultAzureCredential. Producción: establece AZURE_TOKEN_CREDENTIALS=prod o AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# O bien, utiliza una credencial específica directamente en producción:
# Consulta https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

endpoint = "https://.documents.azure.com:443/"

with CosmosClient(url=endpoint, credential=credential) as client:
    # Utiliza el cliente aquí (consulta las secciones siguientes para conocer las operaciones)
    ...

Jerarquía del cliente

Flujo de trabajo principal

Configuración de la base de datos y el contenedor

# Obtener o crear la base de datos
database = client.create_database_if_not_exists(id="mydb")

# Obtener o crear un contenedor con clave de partición
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Obtener los ya existentes
base_de_datos = client.get_database_client("mydb")
contenedor = base_de_datos.get_container_client("mycontainer")

Crear elemento

item = {
    "id": "item-001",           # Obligatorio: único dentro de la partición
    "category": "electronics",   # Valor de la clave de partición
    "name": "Laptop",
    "price": 999.99,
    "tags": ["ordenador", "portátil"]
}

created = container.create_item(body=item)
print(f"Creado: {created['id']}")

Leer elemento

# La lectura requiere el id Y la clave de partición
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Nombre: {item['name']}")

Actualizar elemento (sustituir)

item = container.read_item(item="item-001", partition_key="electronics")
item["price"] = 899.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=item)

Insertar o actualizar un elemento

# Crear si no existe, sustituir si existe
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)

Eliminar elemento

container.delete_item(
    item="item-001",
    partition_key="electronics"
)

Consultas

Consulta básica

# Consulta dentro de una partición (eficiente)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    partition_key="electronics"
)

for item in items:
    print(f"{item['name']}: ${item['price']}")
</code></pre>
<h3>Consulta entre particiones</h3>
<pre><code class="language-python"># Entre particiones (más costosa, utilízala con moderación)
query = "SELECT * FROM c WHERE c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[{"name": "@max_price", "value": 500}],
    enable_cross_partition_query=True
)

for item in items:
    print(item)
</code></pre>
<h3>Consulta con proyección</h3>
<pre><code class="language-python">query = "SELECT c.id, c.name, c.price FROM c WHERE c.category = @category"
items = container.query_items(
    query=query,
    parameters=[{"name": "@category", "value": "electronics"}],
    partition_key="electronics"
)
</code></pre>
<h3>Leer todos los elementos</h3>
<pre><code class="language-python"># Leer todos los elementos de una partición
items = container.read_all_items()  # Entre particiones
# O con clave de partición
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)
</code></pre>
<h2>Claves de partición</h2>
<p><strong>Importante</strong>: Incluye siempre la clave de partición para garantizar la eficiencia de las operaciones.</p>
<pre><code class="language-python">from azure.cosmos import PartitionKey

# Clave de partición única
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Clave de partición jerárquica (versión preliminar)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
</code></pre>
<h2>Rendimiento</h2>
<pre><code class="language-python"># Crear un contenedor con un rendimiento predefinido
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Leer el rendimiento actual
offer = container.read_offer()
print(f"Rendimiento: {offer.offer_throughput} RU/s")

# Actualizar el rendimiento
container.replace_throughput(throughput=1000)
</code></pre>
<h2>Cliente asíncrono</h2>
<pre><code class="language-python">from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential

async def cosmos_operations():
    async with DefaultAzureCredential() as credential:
        async with CosmosClient(endpoint, credential=credential) as client:
            database = client.get_database_client("mydb")
            container = database.get_container_client("mycontainer")
            
            # Crear
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Leer
            item = await container.read_item(item="1", partition_key="test")
            
            # Consulta
            async for item in container.query_items(
                query="SELECT * FROM c",
                partition_key="test"
            ):
                print(item)

import asyncio
asyncio.run(cosmos_operations())
</code></pre>
<h2>Gestión de errores</h2>
<pre><code class="language-python">from azure.cosmos.exceptions import CosmosHttpResponseError

try:
    item = container.read_item(item="nonexistent", partition_key="pk")
except CosmosHttpResponseError as e:
    if e.status_code == 404:
        print("Elemento no encontrado")
    elif e.status_code == 429:
        print(f"Límite de solicitudes alcanzado. Reintentar después de: {e.headers.get('x-ms-retry-after-ms')} ms")
    else:
        raise
</code></pre>
<h2>Prácticas recomendadas</h2>
<ol>
<li><strong>Elige entre sincrónico O asíncrono y mantén la coherencia.</strong> No mezcles los clientes sincrónicos de <code>azure.cosmos</code> con los clientes asíncronos de <code>azure.cosmos.aio</code> en la misma ruta de llamada. Elige un modo por módulo.</li>
<li><strong>Utilice siempre gestores de contexto para los clientes y las credenciales asíncronas.</strong> Envuelva cada cliente en <code>with CosmosClient(...) as client:</code> (síncrono) o <code>async with CosmosClient(...) as client:</code> (asíncrono). Para el modo asíncrono <code>DefaultAzureCredential</code> de <code>azure.identity.aio</code>, utilice también <code>async with credential:</code> para que se eliminen los tokens y los transportes.</li>
<li><strong>Utiliza <code>DefaultAzureCredential</code></strong> para una autenticación portátil entre el entorno de desarrollo local y Azure (evita las cadenas de conexión y las claves de API siempre que sea posible).</li>
<li><strong>Especifica siempre la clave de partición</strong> para lecturas y consultas puntuales</li>
<li><strong>Utiliza consultas parametrizadas</strong> para evitar inyecciones y mejorar el almacenamiento en caché</li>
<li><strong>Evita las consultas entre particiones</strong> siempre que sea posible</li>
<li><strong>Utiliza <code>upsert_item</code></strong> para escrituras idempotentes</li>
<li><strong>Utilice un cliente asíncrono</strong> para escenarios de alto rendimiento</li>
<li><strong>Diseñe la clave de partición</strong> para una distribución uniforme de los datos</li>
<li><strong>Utilice <code>read_item</code></strong> en lugar de una consulta para la recuperación de un único documento</li>
</ol>
<h2>Archivos de referencia</h2>
<table>
<thead>
<tr>
<th>Archivo</th>
<th>Contenido</th>
</tr>
</thead>
<tbody><tr>
<td>references/partitioning.md</td>
<td>Estrategias de claves de partición, claves jerárquicas, detección y mitigación de particiones «calientes»</td>
</tr>
<tr>
<td>references/query-patterns.md</td>
<td>Optimización de consultas, agregaciones, paginación, transacciones y feed de cambios</td>
</tr>
<tr>
<td>scripts/setup_cosmos_container.py</td>
<td>Herramienta de línea de comandos para crear contenedores con partición, rendimiento e indexación</td>
</tr>
</tbody></table>