SDK Azure Cosmos DB pour Python

Bibliothèque cliente pour l'API NoSQL d'Azure Cosmos DB — base de données multimodèle distribuée à l'échelle mondiale.

Installation

pip install azure-cosmos azure-identity

Variables d’environnement

COSMOS_ENDPOINT=https://.documents.azure.com:443/  # Requis pour toutes les méthodes d'authentification
COSMOS_DATABASE=mydb  # Requis pour toutes les méthodes d'authentification
COSMOS_CONTAINER=mycontainer  # Requis pour toutes les méthodes d'authentification
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification et cycle de vie

🔑 Deux règles s’appliquent à tous les exemples de code ci-dessous :

1. Privilégiez DefaultAzureCredential. Il fonctionne en local (Azure CLI / VS Code / Developer CLI) et dans Azure (identité gérée, identité de charge de travail) sans modification du code. Évitez les chaînes de connexion, les identifiants de compte et les clés API : ils contournent l’audit et la rotation Entra.
   • Développement local : DefaultAzureCredential fonctionne tel quel.
   • En production : définissez AZURE_TOKEN_CREDENTIALS=prod (ou AZURE_TOKEN_CREDENTIALS=) pour limiter la chaîne d’informations d’identification aux informations d’identification sécurisées pour la production.
2. Enveloppez chaque client dans un gestionnaire de contexte afin que les transports HTTP, les sockets et les caches de jetons soient libérés de manière déterministe :
   • Synchrone : avec (...) comme client :
   • Asynchrone : async avec (...) comme client : et async avec DefaultAzureCredential() comme identifiant : (de azure.identity.aio)

Les extraits de code peuvent simplifier cette configuration, mais le code de production doit toujours respecter ces deux règles.

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

# Développement local : DefaultAzureCredential. En production : définissez AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# Ou utilisez directement des informations d’identification spécifiques en production :
# Voir 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:
    # Utilisez le client ici (voir les sections suivantes pour les opérations)
    ...

Hiérarchie des clients

Flux de travail principal

Configuration de la base de données et du conteneur

# Récupérer ou créer la base de données
database = client.create_database_if_not_exists(id="mydb")

# Récupérer ou créer un conteneur avec une clé de partition
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Récupérer les entités existantes
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Créer un élément

item = {
    "id": "item-001",           # Obligatoire : unique au sein de la partition
    "category": "electronics",   # Valeur de la clé de partition
    "name": "Laptop",
    "price": 999,99,
    "tags": ["ordinateur", "portable"]
}

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

Lecture d’un élément

# La lecture nécessite l'identifiant ET la clé de partition
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Nom : {item['name']}")

Mise à jour d’un élément (remplacement)

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)

Insérer ou mettre à jour un élément

# Créer si l’élément n’existe pas, remplacer s’il existe
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablette",
    "price": 499,99
}

result = container.upsert_item(body=item)

Supprimer un élément

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

Requêtes

Requête de base

# Requête au sein d'une partition (efficace)
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>Requête inter-partitions</h3>
<pre><code class="language-python"># Inter-partitions (plus coûteux, à utiliser avec parcimonie)
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>Requête avec projection</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>Lire tous les éléments</h3>
<pre><code class="language-python"># Lire tous les éléments d’une partition
items = container.read_all_items()  # Entre partitions
# Ou avec la clé de partition
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)
</code></pre>
<h2>Clés de partition</h2>
<p><strong>Important</strong> : incluez toujours une clé de partition pour garantir l’efficacité des opérations.</p>
<pre><code class="language-python">from azure.cosmos import PartitionKey

# Clé de partition unique
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Clé de partition hiérarchique (version préliminaire)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
</code></pre>
<h2>Débit</h2>
<pre><code class="language-python"># Créer un conteneur avec un débit provisionné
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Lecture du débit actuel
offer = container.read_offer()
print(f"Débit : {offer.offer_throughput} RU/s")

# Mettre à jour le débit
container.replace_throughput(throughput=1000)
</code></pre>
<h2>Client asynchrone</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")
            
            # Créer
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Lire
            item = await container.read_item(item="1", partition_key="test")
            
            # Requête
            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>Gestion des erreurs</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("Élément introuvable")
    elif e.status_code == 429:
        print(f"Limite de débit atteinte. Réessayer après : {e.headers.get('x-ms-retry-after-ms')} ms")
    else:
        raise
</code></pre>
<h2>Meilleures pratiques</h2>
<ol>
<li><strong>Optez pour le mode synchrone OU asynchrone et restez cohérent.</strong> Ne mélangez pas les clients synchrones <code>azure.cosmos</code> avec les clients asynchrones <code>azure.cosmos.aio</code> dans le même chemin d’appel. Choisissez un seul mode par module.</li>
<li><strong>Utilisez toujours des gestionnaires de contexte pour les clients et les informations d’identification asynchrones.</strong> Enveloppez chaque client dans <code>with CosmosClient(...) as client:</code> (synchrone) ou <code>async with CosmosClient(...) as client:</code> (asynchrone). Pour l’asynchrone <code>DefaultAzureCredential</code> de <code>azure.identity.aio</code>, utilisez également <code>async with credential:</code> afin que les jetons et les transports soient nettoyés.</li>
<li><strong>Utilisez <code>DefaultAzureCredential</code></strong> pour une authentification portable entre le développement local et Azure (évitez les chaînes de connexion / clés API dans la mesure du possible).</li>
<li><strong>Spécifiez toujours une clé de partition</strong> pour les lectures ponctuelles et les requêtes</li>
<li><strong>Utilisez des requêtes paramétrées</strong> pour prévenir les injections et améliorer la mise en cache</li>
<li><strong>Évitez les requêtes inter-partitions</strong> dans la mesure du possible</li>
<li><strong>Utilisez <code>upsert_item</code></strong> pour les écritures idempotentes</li>
<li><strong>Utilisez un client asynchrone</strong> pour les scénarios à haut débit</li>
<li><strong>Concevez la clé de partition</strong> pour une répartition homogène des données</li>
<li><strong>Utilisez <code>read_item</code></strong> plutôt qu’une requête pour la récupération d’un seul document</li>
</ol>
<h2>Fichiers de référence</h2>
<table>
<thead>
<tr>
<th>Fichier</th>
<th>Sommaire</th>
</tr>
</thead>
<tbody><tr>
<td>references/partitioning.md</td>
<td>Stratégies de clés de partition, clés hiérarchiques, détection et atténuation des partitions « chaudes »</td>
</tr>
<tr>
<td>references/query-patterns.md</td>
<td>Optimisation des requêtes, agrégations, pagination, transactions, flux de modifications</td>
</tr>
<tr>
<td>scripts/setup_cosmos_container.py</td>
<td>Outil CLI permettant de créer des conteneurs avec partitionnement, débit et indexation</td>
</tr>
</tbody></table>