SDK do Azure Cosmos DB para Python

Biblioteca cliente para a API NoSQL do Azure Cosmos DB — banco de dados multimodelo distribuído globalmente.

Instalação

pip install azure-cosmos azure-identity

Variáveis de ambiente

COSMOS_ENDPOINT=https://.documents.azure.com:443/  # Obrigatório para todos os métodos de autenticação
COSMOS_DATABASE=mydb  # Obrigatório para todos os métodos de autenticação
COSMOS_CONTAINER=mycontainer  # Obrigatório para todos os métodos de autenticação
AZURE_TOKEN_CREDENTIALS=prod # Obrigatório apenas se DefaultAzureCredential for usado em produção

Autenticação e ciclo de vida

🔑 Duas regras se aplicam a todos os exemplos de código abaixo:

1. Dê preferência ao DefaultAzureCredential. Ele funciona localmente (Azure CLI / VS Code / Developer CLI) e no Azure (identidade gerenciada, identidade de carga de trabalho) sem alteração no código. Evite strings de conexão, contas e chaves de API — elas contornam a auditoria e a rotação do Entra.
   • Desenvolvimento local: o DefaultAzureCredential funciona como está.
   • Produção: defina AZURE_TOKEN_CREDENTIALS=prod (ou AZURE_TOKEN_CREDENTIALS=) para restringir a cadeia de credenciais a credenciais seguras para produção.
2. Envolva cada cliente em um gerenciador de contexto para que transportes HTTP, soquetes e caches de tokens sejam liberados de forma determinística:
   • Sincronizado: com (...) como cliente:
   • Assíncrono: async com (...) como cliente: e async com DefaultAzureCredential() como credencial: (de azure.identity.aio)

Os trechos de código podem abreviar essa configuração, mas o código de produção deve sempre seguir ambas as regras.

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

# Desenvolvimento local: DefaultAzureCredential. Produção: defina AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# Ou use uma credencial específica diretamente em produção:
# Consulte 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:
    # Use o cliente aqui (consulte as seções a seguir para operações)
    ...

Hierarquia do cliente

Fluxo de trabalho principal

Configuração do banco de dados e do contêiner

# Obter ou criar banco de dados
database = client.create_database_if_not_exists(id="mydb")

# Obter ou criar o contêiner com a chave de partição
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Obter os já existentes
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Criar item

item = {
    "id": "item-001",           # Obrigatório: único dentro da partição
    "category": "electronics",   # Valor da chave de partição
    "name": "Laptop",
    "price": 999,99,
    "tags": ["computador", "portátil"]
}

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

Ler item

# A leitura requer o id E a chave de partição
item = container.read_item(
    item="item-001",
    partition_key="eletrônicos"
)
print(f"Nome: {item['name']}")

Atualizar item (substituir)

item = container.read_item(item="item-001", partition_key="eletrônicos")
item["preço"] = 899,99
item["em_promoção"] = True

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

Inserir ou atualizar item

# Criar se não existir, substituir se existir
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499,99
}

result = container.upsert_item(body=item)

Excluir item

container.delete_item(
    item="item-001",
    partition_key="eletrônicos"
)

Consultas

Consulta básica

# Consulta dentro de uma partição (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 partições</h3>
<pre><code class="language-python"># Entre partições (mais dispendioso, use com moderação)
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 com projeção</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>Ler todos os itens</h3>
<pre><code class="language-python"># Ler todos em uma partição
items = container.read_all_items()  # Entre partições
# Ou com chave de partição
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)
</code></pre>
<h2>Chaves de partição</h2>
<p><strong>Importante</strong>: Sempre inclua a chave de partição para operações eficientes.</p>
<pre><code class="language-python">from azure.cosmos import PartitionKey

# Chave de partição única
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Chave de partição hierárquica (pré-visualização)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
</code></pre>
<h2>Taxa de transferência</h2>
<pre><code class="language-python"># Criar contêiner com taxa de transferência provisionada
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Ler a taxa de transferência atual
offer = container.read_offer()
print(f"Taxa de transferência: {offer.offer_throughput} RU/s")

# Atualizar a taxa de transferência
container.replace_throughput(throughput=1000)
</code></pre>
<h2>Cliente assí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")
            
            # Criar
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Ler
            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>Tratamento de erros</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("Item não encontrado")
    elif e.status_code == 429:
        print(f"Limite de taxa atingido. Tente novamente após: {e.headers.get('x-ms-retry-after-ms')} ms")
    else:
        raise
</code></pre>
<h2>Práticas recomendadas</h2>
<ol>
<li><strong>Escolha síncrono OU assíncrono e mantenha a consistência.</strong> Não misture clientes síncronos <code>azure.cosmos</code> com clientes assíncronos <code>azure.cosmos.aio</code> no mesmo caminho de chamada. Escolha um modo por módulo.</li>
<li><strong>Sempre use gerenciadores de contexto para clientes e credenciais assíncronas.</strong> Envolva cada cliente em <code>with CosmosClient(...) as client:</code> (síncrono) ou <code>async with CosmosClient(...) as client:</code> (assíncrono). Para o <code>DefaultAzureCredential</code> assíncrono do <code>azure.identity.aio</code>, use também <code>async with credential:</code> para que os tokens e os transportes sejam limpos.</li>
<li><strong>Use <code>DefaultAzureCredential</code></strong> para autenticação portátil entre o ambiente de desenvolvimento local e o Azure (evite strings de conexão/chaves de API sempre que possível).</li>
<li><strong>Sempre especifique a chave de partição</strong> para leituras pontuais e consultas</li>
<li><strong>Use consultas parametrizadas</strong> para evitar injeção e melhorar o armazenamento em cache</li>
<li><strong>Evite consultas entre partições</strong> sempre que possível</li>
<li><strong>Use <code>upsert_item</code></strong> para gravações idempotentes</li>
<li><strong>Use o cliente assíncrono</strong> para cenários de alta taxa de transferência</li>
<li><strong>Projete a chave de partição</strong> para uma distribuição uniforme dos dados</li>
<li><strong>Use <code>read_item</code></strong> em vez de consultas para recuperação de um único documento</li>
</ol>
<h2>Arquivos de referência</h2>
<table>
<thead>
<tr>
<th>Arquivo</th>
<th>Índice</th>
</tr>
</thead>
<tbody><tr>
<td>references/partitioning.md</td>
<td>Estratégias de chave de partição, chaves hierárquicas, detecção e mitigação de partições ativas</td>
</tr>
<tr>
<td>references/query-patterns.md</td>
<td>Otimização de consultas, agregações, paginação, transações, feed de alterações</td>
</tr>
<tr>
<td>scripts/setup_cosmos_container.py</td>
<td>Ferramenta CLI para criar contêineres com particionamento, taxa de transferência e indexação</td>
</tr>
</tbody></table>