azure-cosmos-py
microsoft/skills
使用 Python SDK 在 Azure Cosmos DB NoSQL API 上执行 CRUD 操作、运行查询并管理容器。
...展开全部Azure Cosmos DB Python SDK
Azure Cosmos DB NoSQL API 的客户端库——一种全球分布式、多模型数据库。
安装
pip install azure-cosmos azure-identity
环境变量
COSMOS_ENDPOINT=https://.documents.azure.com:443/ # 所有身份验证方法均需此项
COSMOS_DATABASE=mydb # 所有身份验证方法均需此项
COSMOS_CONTAINER=mycontainer # 所有身份验证方法均需此项
AZURE_TOKEN_CREDENTIALS=prod # 仅当在生产环境中使用 DefaultAzureCredential 时才需要此项
身份验证与生命周期
🔑 以下每个代码示例均遵循两条规则:
- 优先使用
DefaultAzureCredential。它既可在本地(Azure CLI / VS Code / Developer CLI)使用,也可在 Azure 中(托管身份、工作负载身份)使用,且无需修改代码。请避免使用连接字符串、账户/API 密钥——它们会绕过 Entra 的审计和轮换机制。
- 本地开发:
DefaultAzureCredential可直接使用。- 生产环境:将
AZURE_TOKEN_CREDENTIALS设置为prod(或AZURE_TOKEN_CREDENTIALS=),以将凭据链限制为生产环境安全的凭据。- 将每个客户端封装在上下文管理器中,以确保 HTTP 传输、套接字和令牌缓存能以确定性方式释放:
- 同步模式:
使用 ``(...) as client: - 异步:
async with以及(...) as client: async with DefaultAzureCredential() as credential:(来自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:
# 在此处使用客户端(具体操作请参见后续章节)
...
客户端层次结构
| 客户端 | 用途 | 获取自 |
|---|---|---|
CosmosClient |
账户级操作 | 直接实例化 |
DatabaseProxy |
数据库操作 | client.get_database_client() |
ContainerProxy |
容器/项操作 | database.get_container_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)
对项进行Upsert操作
# 若不存在则创建,若存在则替换
item = {
"id": "item-002",
"category": "electronics",
"name": "平板电脑",
"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>azure.identity.aio</code> 的异步 <code>DefaultAzureCredential</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>用于创建具有分区、吞吐量和索引功能的容器的 CLI 工具</td>
</tr>
</tbody></table> ---
name: azure-cosmos-py
description: Perform CRUD operations, run queries, and manage containers on Azure Cosmos DB NoSQL API using the Python SDK.
license: MIT
---
# Azure Cosmos DB SDK for Python
Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.
## Installation
```bash
pip install azure-cosmos azure-identity
```
## Environment Variables
```bash
COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/ # Required for all auth methods
COSMOS_DATABASE=mydb # Required for all auth methods
COSMOS_CONTAINER=mycontainer # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```
## Authentication & Lifecycle
> **🔑 Two rules apply to every code sample below:**
>
> 1. **Prefer `DefaultAzureCredential`.** It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
> - Local dev: `DefaultAzureCredential` works as-is.
> - Production: set `AZURE_TOKEN_CREDENTIALS=prod` (or `AZURE_TOKEN_CREDENTIALS=<specific_credential>`) to constrain the credential chain to production-safe credentials.
> 2. **Wrap every client in a context manager** so HTTP transports, sockets, and token caches are released deterministically:
> - Sync: `with <Client>(...) as client:`
> - Async: `async with <Client>(...) as client:` **and** `async with DefaultAzureCredential() as credential:` (from `azure.identity.aio`)
>
> Snippets may abbreviate this setup, but production code should always follow both rules.
```python
import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.cosmos import CosmosClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
endpoint = "https://<account>.documents.azure.com:443/"
with CosmosClient(url=endpoint, credential=credential) as client:
# Use client here (see following sections for operations)
...
```
## Client Hierarchy
| Client | Purpose | Get From |
|--------|---------|----------|
| `CosmosClient` | Account-level operations | Direct instantiation |
| `DatabaseProxy` | Database operations | `client.get_database_client()` |
| `ContainerProxy` | Container/item operations | `database.get_container_client()` |
## Core Workflow
### Setup Database and Container
```python
# Get or create database
database = client.create_database_if_not_exists(id="mydb")
# Get or create container with partition key
container = database.create_container_if_not_exists(
id="mycontainer",
partition_key=PartitionKey(path="/category")
)
# Get existing
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")
```
### Create Item
```python
item = {
"id": "item-001", # Required: unique within partition
"category": "electronics", # Partition key value
"name": "Laptop",
"price": 999.99,
"tags": ["computer", "portable"]
}
created = container.create_item(body=item)
print(f"Created: {created['id']}")
```
### Read Item
```python
# Read requires id AND partition key
item = container.read_item(
item="item-001",
partition_key="electronics"
)
print(f"Name: {item['name']}")
```
### Update Item (Replace)
```python
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)
```
### Upsert Item
```python
# Create if not exists, replace if exists
item = {
"id": "item-002",
"category": "electronics",
"name": "Tablet",
"price": 499.99
}
result = container.upsert_item(body=item)
```
### Delete Item
```python
container.delete_item(
item="item-001",
partition_key="electronics"
)
```
## Queries
### Basic Query
```python
# Query within a partition (efficient)
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']}")
```
### Cross-Partition Query
```python
# Cross-partition (more expensive, use sparingly)
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)
```
### Query with Projection
```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"
)
```
### Read All Items
```python
# Read all in a partition
items = container.read_all_items() # Cross-partition
# Or with partition key
items = container.query_items(
query="SELECT * FROM c",
partition_key="electronics"
)
```
## Partition Keys
**Critical**: Always include partition key for efficient operations.
```python
from azure.cosmos import PartitionKey
# Single partition key
container = database.create_container_if_not_exists(
id="orders",
partition_key=PartitionKey(path="/customer_id")
)
# Hierarchical partition key (preview)
container = database.create_container_if_not_exists(
id="events",
partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)
```
## Throughput
```python
# Create container with provisioned throughput
container = database.create_container_if_not_exists(
id="mycontainer",
partition_key=PartitionKey(path="/pk"),
offer_throughput=400 # RU/s
)
# Read current throughput
offer = container.read_offer()
print(f"Throughput: {offer.offer_throughput} RU/s")
# Update throughput
container.replace_throughput(throughput=1000)
```
## Async Client
```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")
# Create
await container.create_item(body={"id": "1", "pk": "test"})
# Read
item = await container.read_item(item="1", partition_key="test")
# Query
async for item in container.query_items(
query="SELECT * FROM c",
partition_key="test"
):
print(item)
import asyncio
asyncio.run(cosmos_operations())
```
## Error Handling
```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 not found")
elif e.status_code == 429:
print(f"Rate limited. Retry after: {e.headers.get('x-ms-retry-after-ms')}ms")
else:
raise
```
## Best Practices
1. **Pick sync OR async and stay consistent.** Do not mix `azure.cosmos` sync clients with `azure.cosmos.aio` async clients in the same call path. Choose one mode per module.
2. **Always use context managers for clients and async credentials.** Wrap every client in `with CosmosClient(...) as client:` (sync) or `async with CosmosClient(...) as client:` (async). For async `DefaultAzureCredential` from `azure.identity.aio`, also use `async with credential:` so tokens and transports are cleaned up.
3. **Use `DefaultAzureCredential`** for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
4. **Always specify partition key** for point reads and queries
5. **Use parameterized queries** to prevent injection and improve caching
6. **Avoid cross-partition queries** when possible
7. **Use `upsert_item`** for idempotent writes
8. **Use async client** for high-throughput scenarios
9. **Design partition key** for even data distribution
10. **Use `read_item`** instead of query for single document retrieval
## Reference Files
| File | Contents |
|------|----------|
| [references/partitioning.md](references/partitioning.md) | Partition key strategies, hierarchical keys, hot partition detection and mitigation |
| [references/query-patterns.md](references/query-patterns.md) | Query optimization, aggregations, pagination, transactions, change feed |
| [scripts/setup_cosmos_container.py](scripts/setup_cosmos_container.py) | CLI tool for creating containers with partitioning, throughput, and indexing |





首页
