Azure Cosmos Py

1---
2name: azure-cosmos-py
3description: |
4  Azure Cosmos DB SDK for Python (NoSQL API). Use for document CRUD, queries, containers, and globally distributed data.
5  Triggers: "cosmos db", "CosmosClient", "container", "document", "NoSQL", "partition key".
6package: azure-cosmos
7---
8 
9# Azure Cosmos DB SDK for Python
10 
11Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.
12 
13## Installation
14 
15```bash
16pip install azure-cosmos azure-identity
17```
18 
19## Environment Variables
20 
21```bash
22COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
23COSMOS_DATABASE=mydb
24COSMOS_CONTAINER=mycontainer
25```
26 
27## Authentication
28 
29```python
30from azure.identity import DefaultAzureCredential
31from azure.cosmos import CosmosClient
32 
33credential = DefaultAzureCredential()
34endpoint = "https://<account>.documents.azure.com:443/"
35 
36client = CosmosClient(url=endpoint, credential=credential)
37```
38 
39## Client Hierarchy
40 
41| Client | Purpose | Get From |
42|--------|---------|----------|
43| `CosmosClient` | Account-level operations | Direct instantiation |
44| `DatabaseProxy` | Database operations | `client.get_database_client()` |
45| `ContainerProxy` | Container/item operations | `database.get_container_client()` |
46 
47## Core Workflow
48 
49### Setup Database and Container
50 
51```python
52# Get or create database
53database = client.create_database_if_not_exists(id="mydb")
54 
55# Get or create container with partition key
56container = database.create_container_if_not_exists(
57    id="mycontainer",
58    partition_key=PartitionKey(path="/category")
59)
60 
61# Get existing
62database = client.get_database_client("mydb")
63container = database.get_container_client("mycontainer")
64```
65 
66### Create Item
67 
68```python
69item = {
70    "id": "item-001",           # Required: unique within partition
71    "category": "electronics",   # Partition key value
72    "name": "Laptop",
73    "price": 999.99,
74    "tags": ["computer", "portable"]
75}
76 
77created = container.create_item(body=item)
78print(f"Created: {created['id']}")
79```
80 
81### Read Item
82 
83```python
84# Read requires id AND partition key
85item = container.read_item(
86    item="item-001",
87    partition_key="electronics"
88)
89print(f"Name: {item['name']}")
90```
91 
92### Update Item (Replace)
93 
94```python
95item = container.read_item(item="item-001", partition_key="electronics")
96item["price"] = 899.99
97item["on_sale"] = True
98 
99updated = container.replace_item(item=item["id"], body=item)
100```
101 
102### Upsert Item
103 
104```python
105# Create if not exists, replace if exists
106item = {
107    "id": "item-002",
108    "category": "electronics",
109    "name": "Tablet",
110    "price": 499.99
111}
112 
113result = container.upsert_item(body=item)
114```
115 
116### Delete Item
117 
118```python
119container.delete_item(
120    item="item-001",
121    partition_key="electronics"
122)
123```
124 
125## Queries
126 
127### Basic Query
128 
129```python
130# Query within a partition (efficient)
131query = "SELECT * FROM c WHERE c.price < @max_price"
132items = container.query_items(
133    query=query,
134    parameters=[{"name": "@max_price", "value": 500}],
135    partition_key="electronics"
136)
137 
138for item in items:
139    print(f"{item['name']}: ${item['price']}")
140```
141 
142### Cross-Partition Query
143 
144```python
145# Cross-partition (more expensive, use sparingly)
146query = "SELECT * FROM c WHERE c.price < @max_price"
147items = container.query_items(
148    query=query,
149    parameters=[{"name": "@max_price", "value": 500}],
150    enable_cross_partition_query=True
151)
152 
153for item in items:
154    print(item)
155```
156 
157### Query with Projection
158 
159```python
160query = "SELECT c.id, c.name, c.price FROM c WHERE c.category = @category"
161items = container.query_items(
162    query=query,
163    parameters=[{"name": "@category", "value": "electronics"}],
164    partition_key="electronics"
165)
166```
167 
168### Read All Items
169 
170```python
171# Read all in a partition
172items = container.read_all_items()  # Cross-partition
173# Or with partition key
174items = container.query_items(
175    query="SELECT * FROM c",
176    partition_key="electronics"
177)
178```
179 
180## Partition Keys
181 
182**Critical**: Always include partition key for efficient operations.
183 
184```python
185from azure.cosmos import PartitionKey
186 
187# Single partition key
188container = database.create_container_if_not_exists(
189    id="orders",
190    partition_key=PartitionKey(path="/customer_id")
191)
192 
193# Hierarchical partition key (preview)
194container = database.create_container_if_not_exists(
195    id="events",
196    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
197)
198```
199 
200## Throughput
201 
202```python
203# Create container with provisioned throughput
204container = database.create_container_if_not_exists(
205    id="mycontainer",
206    partition_key=PartitionKey(path="/pk"),
207    offer_throughput=400  # RU/s
208)
209 
210# Read current throughput
211offer = container.read_offer()
212print(f"Throughput: {offer.offer_throughput} RU/s")
213 
214# Update throughput
215container.replace_throughput(throughput=1000)
216```
217 
218## Async Client
219 
220```python
221from azure.cosmos.aio import CosmosClient
222from azure.identity.aio import DefaultAzureCredential
223 
224async def cosmos_operations():
225    credential = DefaultAzureCredential()
226    
227    async with CosmosClient(endpoint, credential=credential) as client:
228        database = client.get_database_client("mydb")
229        container = database.get_container_client("mycontainer")
230        
231        # Create
232        await container.create_item(body={"id": "1", "pk": "test"})
233        
234        # Read
235        item = await container.read_item(item="1", partition_key="test")
236        
237        # Query
238        async for item in container.query_items(
239            query="SELECT * FROM c",
240            partition_key="test"
241        ):
242            print(item)
243 
244import asyncio
245asyncio.run(cosmos_operations())
246```
247 
248## Error Handling
249 
250```python
251from azure.cosmos.exceptions import CosmosHttpResponseError
252 
253try:
254    item = container.read_item(item="nonexistent", partition_key="pk")
255except CosmosHttpResponseError as e:
256    if e.status_code == 404:
257        print("Item not found")
258    elif e.status_code == 429:
259        print(f"Rate limited. Retry after: {e.headers.get('x-ms-retry-after-ms')}ms")
260    else:
261        raise
262```
263 
264## Best Practices
265 
2661. **Always specify partition key** for point reads and queries
2672. **Use parameterized queries** to prevent injection and improve caching
2683. **Avoid cross-partition queries** when possible
2694. **Use `upsert_item`** for idempotent writes
2705. **Use async client** for high-throughput scenarios
2716. **Design partition key** for even data distribution
2727. **Use `read_item`** instead of query for single document retrieval
273 
274## Reference Files
275 
276| File | Contents |
277|------|----------|
278| [references/partitioning.md](references/partitioning.md) | Partition key strategies, hierarchical keys, hot partition detection and mitigation |
279| [references/query-patterns.md](references/query-patterns.md) | Query optimization, aggregations, pagination, transactions, change feed |
280| [scripts/setup_cosmos_container.py](scripts/setup_cosmos_container.py) | CLI tool for creating containers with partitioning, throughput, and indexing |
281