Azure Cosmos DB SDK für Python

Client-Bibliothek für die Azure Cosmos DB NoSQL-API – eine global verteilte Datenbank mit mehreren Modellen.

Installation

pip install azure-cosmos azure-identity

Umgebungsvariablen

COSMOS_ENDPOINT=https://.documents.azure.com:443/  # Erforderlich für alle Authentifizierungsmethoden
COSMOS_DATABASE=mydb  # Erforderlich für alle Authentifizierungsmethoden
COSMOS_CONTAINER=mycontainer  # Erforderlich für alle Authentifizierungsmethoden
AZURE_TOKEN_CREDENTIALS=prod # Nur erforderlich, wenn „DefaultAzureCredential“ in der Produktion verwendet wird

Authentifizierung und Lebenszyklus

🔑 Für alle folgenden Code-Beispiele gelten zwei Regeln:

1. Verwenden Sie vorzugsweise „DefaultAzureCredential“. Es funktioniert lokal (Azure-CLI / VS Code / Entwickler-CLI) und in Azure (verwaltete Identität, Workload-Identität) ohne Codeänderungen. Vermeiden Sie Verbindungszeichenfolgen, Konto- und API-Schlüssel – diese umgehen die Entra-Prüfung und -Rotation.
   • Lokale Entwicklung: „DefaultAzureCredential“ funktioniert ohne weitere Anpassungen.
   • Produktion: Setzen Sie AZURE_TOKEN_CREDENTIALS=prod (oder AZURE_TOKEN_CREDENTIALS=), um die Anmeldeinformationskette auf produktionssichere Anmeldeinformationen zu beschränken.
2. Hüllen Sie jeden Client in einen Kontextmanager, damit HTTP-Transporte, Sockets und Token-Caches deterministisch freigegeben werden:
   • Synchron: mit ` (...)` als Client:
   • Asynchron: async mit (...) als Client: und async mit DefaultAzureCredential() als Anmeldeinformationen: (aus azure.identity.aio)

Code-Schnipsel können diese Konfiguration zwar verkürzen, aber Produktionscode sollte stets beide Regeln befolgen.

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

# Lokale Entwicklung: DefaultAzureCredential. Produktion: Setze AZURE_TOKEN_CREDENTIALS=prod oder AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# Oder verwenden Sie in der Produktion direkt spezifische Anmeldedaten:
# Siehe 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:
    # Verwenden Sie den Client hier (Operationen siehe folgende Abschnitte)
    ...

Client-Hierarchie

Kern-Workflow

Datenbank und Container einrichten

# Datenbank abrufen oder erstellen
database = client.create_database_if_not_exists(id="mydb")

# Container mit Partitionsschlüssel abrufen oder erstellen
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Vorhandene abrufen
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Element anlegen

item = {
    "id": "item-001",           # Erforderlich: innerhalb der Partition eindeutig
    "category": "electronics",   # Wert des Partitionsschlüssels
    "name": "Laptop",
    "price": 999,99,
    "tags": ["Computer", "tragbar"]
}

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

Element lesen

# Zum Lesen sind die ID UND der Partitionsschlüssel erforderlich
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Name: {item['name']}")

Element aktualisieren (ersetzen)

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)

Element einfügen oder aktualisieren

# Anlegen, falls nicht vorhanden; ersetzen, falls vorhanden
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)

Element löschen

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

Abfragen

Einfache Abfrage

# Abfrage innerhalb einer Partition (effizient)
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>Partitionsübergreifende Abfrage</h3>
<pre><code class="language-python"># Partitionsübergreifend (aufwändiger, sparsam einsetzen)
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>Abfrage mit Projektion</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>Alle Elemente lesen</h3>
<pre><code class="language-python"># Alle Elemente in einer Partition lesen
items = container.read_all_items()  # Partitionsübergreifend
# Oder mit Partitionsschlüssel
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)
</code></pre>
<h2>Partitionsschlüssel</h2>
<p><strong>Wichtig</strong>: Geben Sie für effiziente Operationen immer den Partitionsschlüssel an.</p>
<pre><code class="language-python">from azure.cosmos import PartitionKey

# Einzelner Partitionsschlüssel
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Hierarchischer Partitionsschlüssel (Vorschau)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
</code></pre>
<h2>Durchsatz</h2>
<pre><code class="language-python"># Container mit bereitgestelltem Durchsatz erstellen
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Aktuellen Durchsatz auslesen
offer = container.read_offer()
print(f"Durchsatz: {offer.offer_throughput} RU/s")

# Durchsatz aktualisieren
container.replace_throughput(throughput=1000)
</code></pre>
<h2>Asynchroner Client</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")
            
            # Erstellen
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Lesen
            item = await container.read_item(item="1", partition_key="test")
            
            # Abfrage
            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>Fehlerbehandlung</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("Element nicht gefunden")
    elif e.status_code == 429:
        print(f"Ratenbegrenzung. Wiederholung nach: {e.headers.get('x-ms-retry-after-ms')} ms")
    else:
        raise
</code></pre>
<h2>Bewährte Verfahren</h2>
<ol>
<li><strong>Entscheiden Sie sich für „sync“ ODER „async“ und bleiben Sie dabei.</strong> Mischen Sie keine <code>azure.cosmos</code>-Sync-Clients mit <code>azure.cosmos.aio</code>-Async-Clients im selben Aufrufpfad. Wählen Sie pro Modul einen Modus.</li>
<li><strong>Verwenden Sie für Clients und asynchrone Anmeldeinformationen stets Kontextmanager.</strong> Umschließen Sie jeden Client mit <code>with CosmosClient(...) as client:</code> (synchron) oder <code>async with CosmosClient(...) as client:</code> (asynchron). Für asynchrone <code>DefaultAzureCredential</code> aus <code>azure.identity.aio</code> verwenden Sie ebenfalls <code>async with credential:</code>, damit Tokens und Transporte bereinigt werden.</li>
<li><strong>Verwenden Sie <code>DefaultAzureCredential</code></strong> für eine portable Authentifizierung zwischen lokaler Entwicklung und Azure (vermeiden Sie nach Möglichkeit Verbindungszeichenfolgen / API-Schlüssel).</li>
<li><strong>Geben Sie für Punkt-Lesevorgänge und Abfragen immer den Partitionsschlüssel</strong> an.</li>
<li><strong>Verwenden Sie parametrisierte Abfragen</strong>, um Injektionen zu verhindern und das Caching zu verbessern</li>
<li><strong>Vermeiden Sie nach Möglichkeit partitionenübergreifende Abfragen</strong></li>
<li><strong>Verwenden Sie <code>upsert_item</code></strong> für idempotente Schreibvorgänge</li>
<li><strong>Verwenden Sie den asynchronen Client </strong> für Szenarien mit hohem Durchsatz </li>
<li><strong>Entwerfen Sie den Partitionsschlüssel </strong> für eine gleichmäßige Datenverteilung </li>
<li><strong>Verwenden Sie <code>read_item</code></strong> anstelle einer Abfrage zum Abrufen einzelner Dokumente </li>
</ol>
<h2>Referenzdateien</h2>
<table>
<thead>
<tr>
<th>Datei</th>
<th>Inhalt</th>
</tr>
</thead>
<tbody><tr>
<td>references/partitioning.md</td>
<td>Strategien für Partitionsschlüssel, hierarchische Schlüssel, Erkennung und Abmilderung von „Hot Partitions“</td>
</tr>
<tr>
<td>references/query-patterns.md</td>
<td>Abfrageoptimierung, Aggregationen, Paginierung, Transaktionen, Change Feed</td>
</tr>
<tr>
<td>scripts/setup_cosmos_container.py</td>
<td>CLI-Tool zum Erstellen von Containern mit Partitionierung, Durchsatz und Indizierung</td>
</tr>
</tbody></table>