Azure Cosmos Py

作者 microsoft354361d83247MIT收錄於 2026年10月8日更新於 2026年10月8日

Azure Cosmos DB SDK for Python (NoSQL API). Use for document CRUD, queries, containers, and globally distributed data. Triggers: "cosmos db", "CosmosClient", "container", "document", "NoSQL", "partition key".

精選僅含說明Software Development

AI 產生的概覽

指導 Python 開發者使用 Azure Cosmos DB NoSQL SDK 進行文件 CRUD、查詢、分割區與輸送量設定。

功能: 提供 Azure Cosmos DB Python SDK（azure-cosmos）的參考說明與程式碼範例，涵蓋用戶端初始化、使用 DefaultAzureCredential 的身分驗證、資料庫與容器建立、項目的建立/讀取/更新/upsert/刪除、參數化查詢與跨分割區查詢、分割區索引鍵、輸送量管理、非同步用戶端以及錯誤處理。同時指向兩份關於分割區與查詢模式的參考文件。產出為指引與程式碼片段，而非檔案或可執行結果。
適用情境: 適用於撰寫或審查存取 Azure Cosmos DB NoSQL 帳戶的 Python 程式碼時，尤其是文件 CRUD、查詢設計、分割區索引鍵選擇或輸送量設定。也適合在同步與非同步用戶端之間做選擇，或處理 Cosmos 特有錯誤時使用。
執行需求: 需要 azure-cosmos 與 azure-identity Python 套件、Cosmos DB 帳戶端點，以及透過環境變數提供的資料庫與容器名稱，並需要 Azure 認證（DefaultAzureCredential 或受控識別）。需要連線至 Cosmos DB 帳戶的網路存取。不附帶指令碼，僅為說明與參考文件。

Azure Cosmos DB SDK for Python

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

Installation

Environment Variables

bash

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methodsCOSMOS_DATABASE=mydb  # Required for all auth methodsCOSMOS_CONTAINER=mycontainer  # Required for all auth methodsAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methodsCOSMOS_DATABASE=mydb  # Required for all auth methodsCOSMOS_CONTAINER=mycontainer  # Required for all auth methodsAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

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.

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 osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom 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)    ...

import osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom 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 databasedatabase = client.create_database_if_not_exists(id="mydb")
# Get or create container with partition keycontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/category"))
# Get existingdatabase = client.get_database_client("mydb")container = database.get_container_client("mycontainer")

# Get or create databasedatabase = client.create_database_if_not_exists(id="mydb")
# Get or create container with partition keycontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/category"))
# Get existingdatabase = 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']}")

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 keyitem = container.read_item(    item="item-001",    partition_key="electronics")print(f"Name: {item['name']}")

# Read requires id AND partition keyitem = 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.99item["on_sale"] = True
updated = container.replace_item(item=item["id"], body=item)

item = container.read_item(item="item-001", partition_key="electronics")item["price"] = 899.99item["on_sale"] = True
updated = container.replace_item(item=item["id"], body=item)

Upsert Item

python

# Create if not exists, replace if existsitem = {    "id": "item-002",    "category": "electronics",    "name": "Tablet",    "price": 499.99}
result = container.upsert_item(body=item)

# Create if not exists, replace if existsitem = {    "id": "item-002",    "category": "electronics",    "name": "Tablet",    "price": 499.99}
result = container.upsert_item(body=item)

Delete Item

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

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

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

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 partitionitems = container.read_all_items()  # Cross-partition# Or with partition keyitems = container.query_items(    query="SELECT * FROM c",    partition_key="electronics")

# Read all in a partitionitems = container.read_all_items()  # Cross-partition# Or with partition keyitems = 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 keycontainer = 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"]))

from azure.cosmos import PartitionKey
# Single partition keycontainer = 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 throughputcontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/pk"),    offer_throughput=400  # RU/s)
# Read current throughputoffer = container.read_offer()print(f"Throughput: {offer.offer_throughput} RU/s")
# Update throughputcontainer.replace_throughput(throughput=1000)

# Create container with provisioned throughputcontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/pk"),    offer_throughput=400  # RU/s)
# Read current throughputoffer = container.read_offer()print(f"Throughput: {offer.offer_throughput} RU/s")
# Update throughputcontainer.replace_throughput(throughput=1000)

Async Client

python

from azure.cosmos.aio import CosmosClientfrom 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 asyncioasyncio.run(cosmos_operations())

from azure.cosmos.aio import CosmosClientfrom 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 asyncioasyncio.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

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

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.
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.
Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
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

Reference Files

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

來源與署名

來源：microsoft/skills位於.github/plugins/azure-sdk-python/skills/azure-cosmos-py提交354361d

授權條款： MIT

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架

Azure Cosmos Py

Azure Cosmos DB SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Client Hierarchy

Core Workflow

Setup Database and Container

Create Item

Read Item

Update Item (Replace)

Upsert Item

Delete Item

Queries

Basic Query

Cross-Partition Query

Query with Projection

Read All Items

Partition Keys

Throughput

Async Client

Error Handling

Best Practices

Reference Files

來源與署名

更多來自 microsoft/skills 的技能

Customize

Discover Azure Skills

Azure Resource Visualizer

Azure Resource Lookup

Azure Messaging

Azure Kusto

更多Software Development技能

Playground

M5 Onboard

Cardputer Buddy

Testing Strategy

Tech Debt

System Design