Azure Cosmos Py

by microsoft354361d83247MITListed Oct 8, 2026Updated Oct 8, 2026

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

FeaturedInstructions onlySoftware Development

AI-generated overview

Guides Python developers using the Azure Cosmos DB NoSQL SDK for document CRUD, queries, partitioning and throughput.

What it does: Provides reference instructions and code samples for the Azure Cosmos DB Python SDK (azure-cosmos), covering client setup, authentication with DefaultAzureCredential, database and container creation, item create/read/update/upsert/delete, parameterized and cross-partition queries, partition keys, throughput management, async clients and error handling. It also points to two reference documents on partitioning and query patterns. It produces guidance and snippets rather than files or runnable output.
When to use it: Use when writing or reviewing Python code that talks to Azure Cosmos DB NoSQL accounts, especially for document CRUD, query design, partition key choices or throughput settings. Also useful when deciding between sync and async clients or handling Cosmos-specific errors.
Requirements: Requires the azure-cosmos and azure-identity Python packages, a Cosmos DB account endpoint, database and container names via environment variables, and Azure credentials (DefaultAzureCredential or managed identity). Network access to the Cosmos DB account is needed. Ships no scripts; instructions and reference documents only.

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

Source and attribution

Source:microsoft/skillsin.github/plugins/azure-sdk-python/skills/azure-cosmos-pyat commit354361d

License: MIT

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

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

Source and attribution

More from microsoft/skills

Customize

Discover Azure Skills

Azure Resource Visualizer

Azure Resource Lookup

Azure Messaging

Azure Kusto

More in Software Development

Playground

M5 Onboard

Cardputer Buddy

Dpop Adoption

Gemini Live Api

Gemini Api