SDK Azure Cosmos DB для Python

Клиентская библиотека для NoSQL-API Azure Cosmos DB — глобально распределенной мультимодельной базы данных.

Установка

pip install azure-cosmos azure-identity

Переменные среды

COSMOS_ENDPOINT=https://.documents.azure.com:443/  # Требуется для всех методов аутентификации
COSMOS_DATABASE=mydb  # Требуется для всех методов аутентификации
COSMOS_CONTAINER=mycontainer  # Требуется для всех методов аутентификации
AZURE_TOKEN_CREDENTIALS=prod # Требуется только в том случае, если в производственной среде используется DefaultAzureCredential

Аутентификация и жизненный цикл

🔑 К каждому приведённому ниже примеру кода применяются два правила:

1. Предпочтительно использовать DefaultAzureCredential. Он работает как локально (Azure CLI / VS Code / Developer CLI), так и в Azure (управляемая идентичность, идентичность рабочей нагрузки) без изменения кода. Избегайте использования строк подключения, учетных записей и ключей API — они обходят аудит и ротацию Entra.
   • Локальная разработка: DefaultAzureCredential работает без изменений.
   • Производственная среда: установите AZURE_TOKEN_CREDENTIALS=prod (или AZURE_TOKEN_CREDENTIALS=), чтобы ограничить цепочку учетных данных учетными данными, безопасными для производственной среды.
2. Оберните каждый клиент в менеджер контекста, чтобы HTTP-транспорт, сокеты и кэши токенов освобождались детерминированно:
   • Синхронный режим: с (...) в качестве клиента:
   • Асинхронный режим: async с (...) в качестве клиента: и async с DefaultAzureCredential() в качестве учетных данных: (из azure.identity.aio)

В фрагментах кода эту настройку можно сократить, но в производственном коде всегда следует соблюдать оба правила.

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

# Локальная разработка: DefaultAzureCredential. Производственная среда: установите AZURE_TOKEN_CREDENTIALS=prod или AZURE_TOKEN_CREDENTIALS=
credential = DefaultAzureCredential(require_envvar=True)
# Или используйте конкретные учетные данные напрямую в производственной среде:
# См. 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:
    # Используйте клиент здесь (операции см. в следующих разделах)
    ...

Иерархия клиентов

Основной рабочий процесс

Настройка базы данных и контейнера

# Получить или создать базу данных
database = client.create_database_if_not_exists(id="mydb")

# Получить или создать контейнер с ключом разбиения
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Получить существующие объекты
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")

Создать элемент

item = {
    "id": "item-001",           # Обязательно: уникально внутри раздела
    "category": "electronics",   # Значение ключа раздела
    "name": "Laptop",
    "price": 999.99,
    "tags": ["computer", "portable"]
}

created = container.create_item(body=item)
print(f"Создано: {created['id']}")

Чтение элемента

# Для чтения требуется id И ключ раздела
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Название: {item['name']}")

Обновление элемента (замена)

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)

Добавление или обновление элемента

# Создать, если не существует, заменить, если существует
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)

Удаление элемента

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

Запросы

Базовый запрос

# Запрос внутри раздела (эффективный)
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>Межпартиционный запрос</h3>
<pre><code class="language-python"># Межпартиционный (более ресурсоемкий, использовать с осторожностью)
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>Запрос с проекцией</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>Чтение всех элементов</h3>
<pre><code class="language-python"># Чтение всех элементов в одном разделе
items = container.read_all_items()  # Чтение по всем разделам
# Или с использованием ключа раздела
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)
</code></pre>
<h2>Ключи раздела</h2>
<p><strong>Важно</strong>: всегда указывайте ключ раздела для обеспечения эффективности операций.</p>
<pre><code class="language-python">from azure.cosmos import PartitionKey

# Один ключ раздела
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Иерархический ключ раздела (предварительная версия)
container = database.create_container_if_not_exists(
    id="events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
</code></pre>
<h2>Пропускная способность</h2>
<pre><code class="language-python"># Создание контейнера с заданной пропускной способностью
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Чтение текущей пропускной способности
offer = container.read_offer()
print(f"Пропускная способность: {offer.offer_throughput} RU/s")

# Обновление пропускной способности
container.replace_throughput(throughput=1000)
</code></pre>
<h2>Асинхронный клиент</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")
            
            # Создание
            await container.create_item(body={"id": "1", "pk": "test"})
            
            # Чтение
            item = await container.read_item(item="1", partition_key="test")
            
            # Запрос
            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>Обработка ошибок</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("Элемент не найден")
    elif e.status_code == 429:
        print(f"Ограничение частоты запросов. Повторите попытку через: {e.headers.get('x-ms-retry-after-ms')} мс")
    else:
        raise
</code></pre>
<h2>Рекомендации</h2>
<ol>
<li><strong>Выбирайте синхронный ИЛИ асинхронный режим и придерживайтесь его.</strong> Не смешивайте <code>azure.cosmos</code> синхронных клиентов с <code>azure.cosmos.aio</code> асинхронными клиентами в одном пути вызова. Выбирайте один режим для каждого модуля.</li>
<li><strong>Всегда используйте контекстные менеджеры для клиентов и асинхронных учетных данных.</strong> Оборачивайте каждого клиента в <code>with CosmosClient(...) as client:</code> (синхронный режим) или <code>async with CosmosClient(...) as client:</code> (асинхронный режим). Для асинхронного режима <code>DefaultAzureCredential</code> из <code>azure.identity.aio</code> также используйте <code>async with credential:</code>, чтобы обеспечить очистку токенов и транспортных данных.</li>
<li><strong>Используйте <code>DefaultAzureCredential</code></strong> для переносимой аутентификации между локальной средой разработки и Azure (по возможности избегайте строк подключения / ключей API).</li>
<li><strong>Всегда указывайте ключ раздела</strong> для точечных чтений и запросов</li>
<li><strong>Используйте параметризованные запросы</strong> для предотвращения инъекций и улучшения кэширования</li>
<li><strong>По возможности избегайте межпартиционных запросов</strong></li>
<li><strong>Используйте <code>upsert_item</code></strong> для идемпотентной записи</li>
<li><strong>Используйте асинхронный клиент</strong> для сценариев с высокой пропускной способностью</li>
<li><strong>Проектируйте ключ раздела</strong> для равномерного распределения данных</li>
<li><strong>Используйте <code>read_item</code></strong> вместо запроса для извлечения отдельного документа</li>
</ol>
<h2>Справочные файлы</h2>
<table>
<thead>
<tr>
<th>Файл</th>
<th>Содержание</th>
</tr>
</thead>
<tbody><tr>
<td>references/partitioning.md</td>
<td>Стратегии ключей разбиения, иерархические ключи, обнаружение «горячих» разбиений и меры по их устранению</td>
</tr>
<tr>
<td>references/query-patterns.md</td>
<td>Оптимизация запросов, агрегации, пагинация, транзакции, канал изменений</td>
</tr>
<tr>
<td>scripts/setup_cosmos_container.py</td>
<td>Инструмент командной строки для создания контейнеров с разбиением на партиции, пропускной способностью и индексированием</td>
</tr>
</tbody></table>