azure-cosmos-py

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Azure Cosmos DB SDK for Python

Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.

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

Installation

安装

bash

pip install azure-cosmos azure-identity

bash

pip install azure-cosmos azure-identity

Environment Variables

环境变量

bash

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

bash

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

Authentication

身份验证

python

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)

python

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)

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

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

Core Workflow

核心工作流

Setup Database and Container

配置数据库与容器

python

undefined

python

undefined

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

undefined

database = client.get_database_client("mydb") container = database.get_container_client("mycontainer")

undefined

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

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

undefined

python

undefined

Read requires id AND partition key

item = container.read_item( item="item-001", partition_key="electronics" ) print(f"Name: {item['name']}")

undefined

item = container.read_item( item="item-001", partition_key="electronics" ) print(f"Name: {item['name']}")

undefined

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)

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

插入更新项（Upsert）

python

undefined

python

undefined

Create if not exists, replace if exists

item = { "id": "item-002", "category": "electronics", "name": "Tablet", "price": 499.99 }

result = container.upsert_item(body=item)

undefined

item = { "id": "item-002", "category": "electronics", "name": "Tablet", "price": 499.99 }

result = container.upsert_item(body=item)

undefined

Delete Item

删除项

python

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

python

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

Queries

查询

Basic Query

基础查询

python

undefined

python

undefined

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

undefined

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

undefined

Cross-Partition Query

跨分区查询

python

undefined

python

undefined

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)

undefined

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)

undefined

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

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

undefined

python

undefined

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

undefined

items = container.query_items( query="SELECT * FROM c", partition_key="electronics" )

undefined

Partition Keys

分区键

Critical: Always include partition key for efficient operations.

python

from azure.cosmos import PartitionKey

关键提示：执行高效操作时务必指定分区键。

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

undefined

container = database.create_container_if_not_exists( id="events", partition_key=PartitionKey(path=["/tenant_id", "/user_id"]) )

undefined

Throughput

吞吐量

python

undefined

python

undefined

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)

undefined

container.replace_throughput(throughput=1000)

undefined

Async Client

异步客户端

python

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

python

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

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

最佳实践

Always specify partition key for point reads and queries
Use parameterized queries to prevent injection and improve caching
Avoid cross-partition queries when possible
Use
upsert_item
for idempotent writes
Use async client for high-throughput scenarios
Design partition key for even data distribution
Use
read_item
instead of query for single document retrieval

始终指定分区键，用于点读和查询操作
使用参数化查询，防止注入攻击并提升缓存效率
尽可能避免跨分区查询
**使用
```
upsert_item
```
**实现幂等写入
高吞吐量场景下使用异步客户端
设计分区键以实现数据均匀分布
**读取单个文档时使用
```
read_item
```
**而非查询

Reference Files

参考文件

File	Contents
references/partitioning.md	Partition key strategies, hierarchical keys, hot partition detection and mitigation
references/query-patterns.md	Query optimization, aggregations, pagination, transactions, change feed
scripts/setup_cosmos_container.py	CLI tool for creating containers with partitioning, throughput, and indexing

文件	内容
references/partitioning.md	分区键策略、分层键、热点分区检测与缓解方案
references/query-patterns.md	查询优化、聚合、分页、事务、变更源
scripts/setup_cosmos_container.py	用于创建容器的CLI工具，支持分区、吞吐量和索引配置