azure-cosmos-py

捆绑的上游支持文件

支持文件位于相对于技能目录的路径下。有关其固定来源、许可证、哈希值和导入变更，请参阅 UPSTREAM-NOTICE.md。

支持材料可用性

此目录副本中以下路径仍不可用：

• <support-root>/scripts/setup_cosmos_container.py

请勿执行需要这些文件的步骤。使用前，请获取原始材料并审查其依赖项、许可证和影响。在用户授权范围内，可以继续执行独立的内联步骤和捆绑的辅助工具。

Azure Cosmos DB SDK for Python

适用于 Azure Cosmos DB NoSQL API 的客户端库——全球分布式多模型数据库。

安装

pip install azure-cosmos azure-identity

环境变量

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
COSMOS_DATABASE=mydb
COSMOS_CONTAINER=mycontainer

身份验证

from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient

credential = DefaultAzureCredential()
endpoint = "https://<account>.documents.azure.com:443/"

client = CosmosClient(url=endpoint, credential=credential)

客户端层级

客户端	用途	获取方式
CosmosClient	帐户级操作	直接实例化
DatabaseProxy	数据库操作	client.get_database_client()
ContainerProxy	容器/项目操作	database.get_container_client()

核心工作流

在预配或更改吞吐量之前，请确认帐户、数据库、容器、分区键定义和授权预算。create_*_if_not_exists 不会协调现有容器配置。示例使用不同的键路径；请勿将它们用于同一个容器。租户分区不等于授权隔离。对于读取-修改-写入操作，请使用 ETag 条件；对于重试，请明确执行逻辑操作去重。更改馈送不是全局有序且恰好执行一次的账本；只有在成功完成幂等处理后才进行检查点确认，并核实固定版本 SDK 的检查点 API。

设置数据库和容器

from azure.cosmos import PartitionKey

# 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")

创建项目

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 requires id AND partition key
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Name: {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 项目

# Create if not exists, replace if exists
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 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 (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 = "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 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"
)

分区键

重要：始终包含分区键，以提高操作效率。

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"])
)

吞吐量

# 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.get_throughput()
print(f"Throughput: {offer.offer_throughput} RU/s")

# Update throughput
container.replace_throughput(throughput=1000)

异步客户端

from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential

async def cosmos_operations():
    async with DefaultAzureCredential() as credential, 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())

错误处理

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

最佳实践

1. 点读取和查询始终指定分区键
2. 使用参数化查询以防止注入并改善缓存
3. 尽可能避免跨分区查询
4. 谨慎使用稳定 ID；upsert 会替换现有内容，且不能保证并发控制或逻辑操作去重
5. 在高吞吐量场景使用异步客户端
6. 设计分区键以实现数据均匀分布
7. 使用 read_item 而不是查询来检索单个文档

参考文件

文件	内容
references/partitioning.md	分区键策略、分层键、热分区检测与缓解
references/query-patterns.md	查询优化、聚合、分页、事务、更改馈送
scripts/setup_cosmos_container.py	用于创建容器并设置分区、吞吐量和索引的命令行工具

适用场景

当需要执行概述中描述的工作流或操作时，可以使用此技能。

限制

• 仅当任务明确符合上述范围时才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少所需输入、权限、安全边界或成功标准，请停止并请求澄清。