选项
首页首页 Skill 数据库管理 azure-cosmos-py

azure-cosmos-py

microsoft/skills microsoft/skills

使用 Python SDK 在 Azure Cosmos DB NoSQL API 上执行 CRUD 操作、运行查询并管理容器。

...展开全部
12
更新时间 2026-09-12

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 时才需要此项

身份验证与生命周期

🔑 以下每个代码示例均遵循两条规则:

  1. 优先使用DefaultAzureCredential它既可在本地(Azure CLI / VS Code / Developer CLI)使用,也可在 Azure 中(托管身份、工作负载身份)使用,且无需修改代码。请避免使用连接字符串、账户/API 密钥——它们会绕过 Entra 的审计和轮换机制。
    • 本地开发:DefaultAzureCredential可直接使用。
    • 生产环境:将AZURE_TOKEN_CREDENTIALS设置为prod(或AZURE_TOKEN_CREDENTIALS= ),以将凭据链限制为生产环境安全的凭据。
  2. 将每个客户端封装在上下文管理器中,以确保 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>                                
在 GitHub 上查看
---
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 |

所有文件

0 个文件

安装 azure-cosmos-py

下载技能文件并将其解压到 .claude/skills/ 目录中。

下载ZIP

克隆仓库并复制技能文件到您的项目中。

git clone https://github.com/microsoft/skills/tree/main/.github/plugins/azure-sdk-python/skills/azure-cosmos-py # Copy SKILL.md to your .claude/skills/ directory

复制 复制
快速设置: 将技能文件夹复制到 .claude/skills/ Claude 会自动检测并使用该技能

相关技能

microservices-patterns
更新时间 2026-06-29
jpa-patterns
更新时间 2026-06-30
fabric-lakehouse
更新时间 2026-06-30
prisma-expert
更新时间 2026-06-29