---

name: azure-cosmos-py description: Azure Cosmos DB Python SDK（NoSQL API）。用于文档 CRUD、查询、容器管理和全球分布式数据。 risk: unknown source: community date_added: '2026-02-27'

Azure Cosmos DB Python SDK

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() |

核心工作流

设置数据库和容器

# 获取或创建数据库
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: {created['id']}")

读取项目

# 读取需要 id 和分区键
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 项目

# 不存在则创建，存在则替换
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']}")

跨分区查询

# 跨分区（开销更大，谨慎使用）
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"
)

读取所有项目

# 读取分区内所有项目
items = container.read_all_items()  # 跨分区
# 或使用分区键
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)

分区键

关键：始终包含分区键以实现高效操作。

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

吞吐量

# 创建带预配置吞吐量的容器
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"Throughput: {offer.offer_throughput} RU/s")

# 更新吞吐量
container.replace_throughput(throughput=1000)

异步客户端

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

async def cosmos_operations():
    credential = DefaultAzureCredential()

    async with CosmosClient(endpoint, credential=credential) as client:
        database = client.get_database_client("mydb")
        container = database.get_container_client("mycontainer")

        item = await container.read_item(item="item-001", partition_key="electronics")
        print(item)