azure-cosmos-py

✓ Official★ 2,700

by microsoft · part of microsoft/skills

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

🔥🔥🔥✓ VerifiedFreeQuick setup

#automation #database #cloud #git

🧩 One of 7 skills in the microsoft/skills package — works on its own, and pairs well with its siblings.

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

Inspect the full instructions your agent will receiveExpand

This is the exact playbook injected into your agent when the skill activates — shown here so you can audit it before installing. You don't need to read it to use the skill.

by microsoft

Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database. npx skills add https://github.com/microsoft/agent-skills --skill azure-cosmos-py Download ZIP GitHub2.7k

Azure Cosmos DB SDK for Python

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

Environment Variables

Copy & paste — that's it

COSMOS_ENDPOINT=https:// .documents.azure.com:443/ # Required for all auth methods
COSMOS_DATABASE=mydb # Required for all auth methods
COSMOS_CONTAINER=mycontainer # Required for all auth methods
AZURE_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.

Copy & paste — that's it

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.cosmos import CosmosClient

# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS= 
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:// .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

Copy & paste — that's it

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

Create Item

Copy & paste — that's it

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

Copy & paste — that's it

# Read requires id AND partition key
item = container.read_item(
 item="item-001",
 partition_key="electronics"
)
print(f"Name: {item['name']}")

Update Item (Replace)

Copy & paste — that's it

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

Copy & paste — that's it

# Create if not exists, replace if exists
item = {
 "id": "item-002",
 "category": "electronics",
 "name": "Tablet",
 "price": 499.99
}

result = container.upsert_item(body=item)

Delete Item

Copy & paste — that's it

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

Queries

Basic Query

Copy & paste — that's it

# Query within a partition (efficient)
query = "SELECT * FROM c WHERE c.price **Critical**: Always include partition key for efficient operations.

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

Copy & paste — that's it


## Throughput

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)

Copy & paste — that's it


## Async Client

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

Copy & paste — that's it


## Error Handling

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

Copy & paste — that's it


## 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](https://github.com/microsoft/agent-skills/blob/main/.github/plugins/azure-sdk-python/skills/azure-cosmos-py/references/partitioning.md) Partition key strategies, hierarchical keys, hot partition detection and mitigation 
 [references/query-patterns.md](https://github.com/microsoft/agent-skills/blob/main/.github/plugins/azure-sdk-python/skills/azure-cosmos-py/references/query-patterns.md) Query optimization, aggregations, pagination, transactions, change feed 
 [scripts/setup_cosmos_container.py](https://github.com/microsoft/agent-skills/blob/main/.github/plugins/azure-sdk-python/skills/azure-cosmos-py/scripts/setup_cosmos_container.py) CLI tool for creating containers with partitioning, throughput, and indexing

Copy & paste — that's it

pip install azure-cosmos azure-identity

Run this in your project — your agent picks the skill up automatically.

Installation

Copy & paste — that's it

pip install azure-cosmos azure-identity