# SkillPatch skill: azure-cosmos-py

This skill provides comprehensive guidance for using the Azure Cosmos DB SDK for Python (NoSQL API). It covers authentication best practices using DefaultAzureCredential, client hierarchy setup, and core CRUD operations including creating, reading, and updating documents in Cosmos DB containers. The skill includes environment configuration, installation instructions, and production-ready code patterns.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/azure-cosmos-py
curl -sSL https://skillpatch.dev/install_skill/azure-cosmos-py | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (4)

- `SKILL.md`
- `references/partitioning.md`
- `references/query-patterns.md`
- `scripts/setup_cosmos_container.py`


### `SKILL.md`

````markdown
---
name: azure-cosmos-py
description: |
  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".
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: azure-cosmos
---

# Azure Cosmos DB SDK for Python

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

## Installation

```bash
pip install azure-cosmos azure-identity
```

## Environment Variables

```bash
COSMOS_ENDPOINT=https://<account>.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:**
>
> 1. **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.
> 2. **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 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=<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 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

```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
# Read requires id AND partition key
item = 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.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=item)
```

### Upsert Item

```python
# 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

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

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

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

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

### Read All Items

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

## Partition Keys

**Critical**: Always include partition key for efficient operations.

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

## Throughput

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

## Async Client

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

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

## Best Practices

1. **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.
2. **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.
3. **Use `DefaultAzureCredential`** for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
4. **Always specify partition key** for point reads and queries
5. **Use parameterized queries** to prevent injection and improve caching
6. **Avoid cross-partition queries** when possible
7. **Use `upsert_item`** for idempotent writes
8. **Use async client** for high-throughput scenarios
9. **Design partition key** for even data distribution
10. **Use `read_item`** instead of query for single document retrieval

## Reference Files

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

````


### `references/partitioning.md`

````markdown
# Partition Key Strategies

Comprehensive guide to partition key design and optimization for Azure Cosmos DB.

## Partition Key Fundamentals

### What is a Partition Key?

A partition key determines how data is distributed across physical partitions:

- **Logical partition**: All items with the same partition key value
- **Physical partition**: Storage unit that holds one or more logical partitions
- **Maximum logical partition size**: 20 GB

```python
from azure.cosmos import PartitionKey

# Define partition key at container creation
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)
```

## Choosing a Partition Key

### Good Partition Key Characteristics

| Characteristic | Why It Matters |
|---------------|----------------|
| High cardinality | Many unique values = even distribution |
| Even distribution | Prevents hot partitions |
| Frequently used in queries | Enables efficient single-partition queries |
| Immutable | Partition key cannot be changed after item creation |

### Common Partition Key Patterns

```python
# E-commerce orders - partition by customer
container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)
# Query: "Get all orders for customer X" - efficient single partition

# IoT telemetry - partition by device
container = database.create_container_if_not_exists(
    id="telemetry",
    partition_key=PartitionKey(path="/device_id")
)
# Query: "Get recent readings for device X" - efficient

# Multi-tenant SaaS - partition by tenant
container = database.create_container_if_not_exists(
    id="data",
    partition_key=PartitionKey(path="/tenant_id")
)
# Query: "Get all data for tenant X" - efficient, isolated
```

### Anti-Patterns to Avoid

```python
# BAD: Low cardinality partition key
# Only 2 values = all data in 2 partitions
partition_key=PartitionKey(path="/status")  # "active" or "inactive"

# BAD: Timestamp as partition key (unless with synthetic key)
# Creates write-only partitions, recent partition is always hot
partition_key=PartitionKey(path="/created_date")

# BAD: Monotonically increasing ID
# All recent writes go to the same partition
partition_key=PartitionKey(path="/auto_increment_id")
```

## Hierarchical Partition Keys

For workloads needing multi-level distribution:

```python
from azure.cosmos import PartitionKey

# Two-level hierarchy: tenant → user
container = database.create_container_if_not_exists(
    id="user_events",
    partition_key=PartitionKey(path=["/tenant_id", "/user_id"])
)

# Three-level hierarchy: tenant → department → user
container = database.create_container_if_not_exists(
    id="enterprise_data",
    partition_key=PartitionKey(path=["/tenant_id", "/department", "/user_id"])
)
```

### Querying with Hierarchical Keys

```python
# Full key path - most efficient
items = container.query_items(
    query="SELECT * FROM c WHERE c.status = 'active'",
    partition_key=["tenant-123", "user-456"]  # List for hierarchical
)

# Partial key path - queries within tenant
items = container.query_items(
    query="SELECT * FROM c WHERE c.department = 'engineering'",
    partition_key=["tenant-123"]  # Queries all users in tenant
)
```

### Hierarchical Key Benefits

| Scenario | Single Key | Hierarchical Key |
|----------|------------|------------------|
| Multi-tenant with user isolation | Cross-partition for tenant-wide queries | Single partition for tenant, sub-partitions for users |
| Geographic + temporal data | Choose one dimension | Partition by region, then by date |
| Hot partition mitigation | Limited options | Add synthetic suffix as sub-key |

## Synthetic Partition Keys

When natural keys have poor distribution:

```python
import hashlib

def create_item_with_synthetic_key(container, item: dict, num_buckets: int = 10):
    """Add synthetic partition key for better distribution."""
    # Generate bucket from item ID
    bucket = int(hashlib.md5(item["id"].encode()).hexdigest(), 16) % num_buckets
    
    # Combine natural key with bucket
    item["pk"] = f"{item['category']}_{bucket}"
    
    return container.create_item(body=item)

# Example: Distribute "electronics" category across 10 partitions
item = {
    "id": "item-001",
    "category": "electronics",
    "name": "Laptop"
}
create_item_with_synthetic_key(container, item)
# Results in pk: "electronics_7" (depending on hash)
```

### Time-Based Synthetic Keys

```python
from datetime import datetime

def create_event_with_time_bucket(container, event: dict):
    """Partition by device with hourly buckets."""
    timestamp = event.get("timestamp", datetime.utcnow())
    hour_bucket = timestamp.strftime("%Y%m%d%H")
    
    event["pk"] = f"{event['device_id']}_{hour_bucket}"
    
    return container.create_item(body=event)

# Distributes writes across time-based partitions
event = {
    "id": "event-001",
    "device_id": "device-123",
    "timestamp": datetime.utcnow(),
    "value": 42.5
}
create_event_with_time_bucket(container, event)
# Results in pk: "device-123_2025012812"
```

## Hot Partition Detection

### Monitoring Partition Metrics

```python
from azure.cosmos import CosmosClient

def check_partition_metrics(container):
    """Check container throughput and partition distribution."""
    # Get container properties
    properties = container.read()
    print(f"Container ID: {properties['id']}")
    
    # Get current throughput
    try:
        offer = container.read_offer()
        print(f"Provisioned throughput: {offer.offer_throughput} RU/s")
    except Exception:
        print("Using serverless or database-level throughput")
    
    # Query to check partition key distribution
    query = """
    SELECT COUNT(1) as count, c.pk 
    FROM c 
    GROUP BY c.pk
    """
    
    partition_counts = list(container.query_items(
        query=query,
        enable_cross_partition_query=True
    ))
    
    if partition_counts:
        total = sum(p['count'] for p in partition_counts)
        print(f"\nPartition distribution ({len(partition_counts)} partitions):")
        for p in sorted(partition_counts, key=lambda x: x['count'], reverse=True)[:10]:
            pct = (p['count'] / total) * 100
            print(f"  {p['pk']}: {p['count']} items ({pct:.1f}%)")
            if pct > 20:
                print(f"    ⚠️  Hot partition detected!")
```

### Identifying Hot Partitions via Azure Portal

Key metrics to monitor:
- **Normalized RU Consumption**: >70% indicates potential hot partition
- **Partition Key Range Statistics**: Uneven distribution warnings
- **Throttled Requests (429s)**: May indicate hot partition

### Mitigation Strategies

```python
# Strategy 1: Add randomized suffix
def add_random_suffix(pk: str, num_suffixes: int = 5) -> str:
    """Distribute writes across multiple logical partitions."""
    import random
    suffix = random.randint(0, num_suffixes - 1)
    return f"{pk}_{suffix}"

# Strategy 2: Time-based bucketing
def add_time_bucket(pk: str, bucket_minutes: int = 60) -> str:
    """Create time-based sub-partitions."""
    from datetime import datetime
    bucket = datetime.utcnow().hour
    return f"{pk}_{bucket}"

# Strategy 3: Hash-based distribution
def add_hash_bucket(pk: str, item_id: str, num_buckets: int = 10) -> str:
    """Deterministic distribution based on item ID."""
    import hashlib
    bucket = int(hashlib.sha256(item_id.encode()).hexdigest(), 16) % num_buckets
    return f"{pk}_{bucket}"
```

## Partition Key Design Patterns by Use Case

### E-commerce

```python
# Orders container: partition by customer for order history queries
orders_container = database.create_container_if_not_exists(
    id="orders",
    partition_key=PartitionKey(path="/customer_id")
)

# Products container: partition by category for browsing
products_container = database.create_container_if_not_exists(
    id="products",
    partition_key=PartitionKey(path="/category")
)

# Cart container: partition by session for cart operations
cart_container = database.create_container_if_not_exists(
    id="carts",
    partition_key=PartitionKey(path="/session_id")
)
```

### Multi-tenant SaaS

```python
# Hierarchical: tenant → resource type for isolation + efficient queries
container = database.create_container_if_not_exists(
    id="tenant_data",
    partition_key=PartitionKey(path=["/tenant_id", "/resource_type"])
)

# Document structure
document = {
    "id": "doc-001",
    "tenant_id": "tenant-abc",
    "resource_type": "project",
    "name": "My Project",
    "data": {...}
}
```

### IoT / Time Series

```python
# Hierarchical: device → time bucket for efficient range queries
container = database.create_container_if_not_exists(
    id="telemetry",
    partition_key=PartitionKey(path=["/device_id", "/day"])
)

# Document structure
telemetry = {
    "id": "reading-001",
    "device_id": "sensor-123",
    "day": "2025-01-28",  # Daily bucket
    "timestamp": "2025-01-28T12:30:00Z",
    "temperature": 22.5,
    "humidity": 45.2
}
```

### Social / Activity Feeds

```python
# Partition by user for personal feed queries
container = database.create_container_if_not_exists(
    id="activity_feed",
    partition_key=PartitionKey(path="/user_id")
)

# With write amplification for follower feeds
# Store activity in both author and follower partitions
async def post_activity(container, author_id: str, follower_ids: list[str], activity: dict):
    """Fan-out activity to author and all followers."""
    tasks = []
    
    # Author's partition
    author_activity = {**activity, "user_id": author_id, "id": f"{activity['id']}_author"}
    tasks.append(container.create_item(body=author_activity))
    
    # Each follower's partition
    for follower_id in follower_ids:
        follower_activity = {**activity, "user_id": follower_id, "id": f"{activity['id']}_{follower_id}"}
        tasks.append(container.create_item(body=follower_activity))
    
    await asyncio.gather(*tasks)
```

## Best Practices Summary

| Practice | Recommendation |
|----------|----------------|
| Cardinality | Aim for thousands+ unique values |
| Distribution | Monitor and rebalance if >20% in single partition |
| Query alignment | Partition key should match most frequent query filter |
| Size monitoring | Keep logical partitions under 15GB (20GB limit) |
| Throughput | Consider hierarchical keys for mixed workloads |
| Immutability | Plan partition key carefully - cannot change later |

````


### `references/query-patterns.md`

````markdown
# Query Patterns Reference

Advanced query patterns for Azure Cosmos DB NoSQL API.

## Parameterized Queries

Always use parameters to prevent injection and enable query plan caching:

```python
# GOOD: Parameterized query
query = "SELECT * FROM c WHERE c.category = @category AND c.price < @max_price"
items = container.query_items(
    query=query,
    parameters=[
        {"name": "@category", "value": "electronics"},
        {"name": "@max_price", "value": 500}
    ],
    partition_key="electronics"
)

# BAD: String interpolation (injection risk, no caching)
category = "electronics"
query = f"SELECT * FROM c WHERE c.category = '{category}'"  # Never do this!
```

## Query Optimization

### Single Partition vs Cross-Partition

```python
# EFFICIENT: Single partition query (always preferred)
items = container.query_items(
    query="SELECT * FROM c WHERE c.status = 'active'",
    partition_key="tenant-123"
)

# EXPENSIVE: Cross-partition query (use sparingly)
items = container.query_items(
    query="SELECT * FROM c WHERE c.status = 'active'",
    enable_cross_partition_query=True  # Fans out to all partitions
)
```

### Point Reads vs Queries

```python
# MOST EFFICIENT: Point read (1 RU for 1KB document)
item = container.read_item(
    item="doc-123",
    partition_key="tenant-abc"
)

# LESS EFFICIENT: Query for single document (higher RU)
items = list(container.query_items(
    query="SELECT * FROM c WHERE c.id = @id",
    parameters=[{"name": "@id", "value": "doc-123"}],
    partition_key="tenant-abc"
))
item = items[0] if items else None
```

### Projection for Efficiency

```python
# GOOD: Select only needed fields (less RU, less bandwidth)
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"
)

# EXPENSIVE: Select all fields when you only need a few
query = "SELECT * FROM c WHERE c.category = @category"
```

## Pagination

### Continuation Token Pagination

```python
def paginated_query(
    container,
    query: str,
    partition_key: str,
    page_size: int = 100,
    parameters: list | None = None
):
    """Paginate through query results using continuation tokens."""
    continuation_token = None
    page_number = 0
    
    while True:
        # Create query iterable with pagination
        query_iterable = container.query_items(
            query=query,
            parameters=parameters or [],
            partition_key=partition_key,
            max_item_count=page_size,
            continuation_token=continuation_token
        )
        
        # Get page of results
        page = list(query_iterable.by_page().__next__())
        page_number += 1
        
        print(f"Page {page_number}: {len(page)} items")
        yield page
        
        # Get continuation token for next page
        continuation_token = query_iterable.continuation_token
        if not continuation_token:
            break

# Usage
for page in paginated_query(
    container,
    query="SELECT * FROM c WHERE c.status = 'active'",
    partition_key="tenant-123",
    page_size=50
):
    process_page(page)
```

### Offset-Limit Pagination (Not Recommended for Large Datasets)

```python
# Works but inefficient for large offsets (still reads skipped items)
query = "SELECT * FROM c ORDER BY c.created_at DESC OFFSET @offset LIMIT @limit"
items = container.query_items(
    query=query,
    parameters=[
        {"name": "@offset", "value": 100},
        {"name": "@limit", "value": 10}
    ],
    partition_key="tenant-123"
)
```

## Aggregations

### COUNT, SUM, AVG, MIN, MAX

```python
# Count items in partition
query = "SELECT VALUE COUNT(1) FROM c WHERE c.status = 'active'"
result = list(container.query_items(
    query=query,
    partition_key="tenant-123"
))
count = result[0]  # Returns integer directly with VALUE

# Sum with grouping
query = """
SELECT c.category, SUM(c.quantity) as total_quantity, AVG(c.price) as avg_price
FROM c
WHERE c.order_date >= @start_date
GROUP BY c.category
"""
results = container.query_items(
    query=query,
    parameters=[{"name": "@start_date", "value": "2025-01-01"}],
    partition_key="tenant-123"
)

for result in results:
    print(f"{result['category']}: {result['total_quantity']} items, ${result['avg_price']:.2f} avg")
```

### Aggregate Functions Reference

| Function | Description | Example |
|----------|-------------|---------|
| `COUNT(expr)` | Count non-null values | `COUNT(c.id)` |
| `COUNT(1)` | Count all documents | `COUNT(1)` |
| `SUM(expr)` | Sum numeric values | `SUM(c.quantity)` |
| `AVG(expr)` | Average of numeric values | `AVG(c.price)` |
| `MIN(expr)` | Minimum value | `MIN(c.created_at)` |
| `MAX(expr)` | Maximum value | `MAX(c.price)` |

## Advanced Query Patterns

### Array Operations

```python
# Check if array contains value
query = "SELECT * FROM c WHERE ARRAY_CONTAINS(c.tags, 'urgent')"

# Check if array contains object with property
query = """
SELECT * FROM c 
WHERE ARRAY_CONTAINS(c.items, {'product_id': @product_id}, true)
"""

# Flatten array with JOIN
query = """
SELECT c.id, c.order_id, item.product_name, item.quantity
FROM c
JOIN item IN c.items
WHERE item.quantity > @min_quantity
"""

# Array length
query = "SELECT * FROM c WHERE ARRAY_LENGTH(c.tags) > 3"
```

### String Functions

```python
# Case-insensitive search (expensive - no index)
query = "SELECT * FROM c WHERE LOWER(c.name) = LOWER(@search_name)"

# Contains (prefix search uses index, contains doesn't)
query = "SELECT * FROM c WHERE STARTSWITH(c.name, @prefix)"  # Uses index
query = "SELECT * FROM c WHERE CONTAINS(c.name, @substring)"  # Full scan

# String manipulation
query = """
SELECT 
    c.id,
    CONCAT(c.first_name, ' ', c.last_name) as full_name,
    LENGTH(c.description) as desc_length
FROM c
"""
```

### Subqueries

```python
# EXISTS subquery
query = """
SELECT * FROM c 
WHERE EXISTS(
    SELECT VALUE item FROM item IN c.items WHERE item.status = 'pending'
)
"""

# Scalar subquery
query = """
SELECT 
    c.id,
    c.name,
    (SELECT VALUE COUNT(1) FROM item IN c.items) as item_count
FROM c
"""
```

### Ordering and Distinct

```python
# Order by (requires composite index for multiple fields)
query = "SELECT * FROM c ORDER BY c.created_at DESC, c.priority ASC"

# Distinct values
query = "SELECT DISTINCT VALUE c.category FROM c"

# Top N
query = "SELECT TOP 10 * FROM c ORDER BY c.score DESC"
```

## Transactions (Transactional Batch)

Execute multiple operations atomically within a single partition:

```python
from azure.cosmos import TransactionalBatch

def transfer_funds(container, from_account_id: str, to_account_id: str, amount: float, partition_key: str):
    """Transfer funds between accounts atomically."""
    
    # Read current balances
    from_account = container.read_item(item=from_account_id, partition_key=partition_key)
    to_account = container.read_item(item=to_account_id, partition_key=partition_key)
    
    if from_account["balance"] < amount:
        raise ValueError("Insufficient funds")
    
    # Update balances
    from_account["balance"] -= amount
    to_account["balance"] += amount
    
    # Execute as transaction
    batch = TransactionalBatch(partition_key=partition_key)
    batch.replace_item(item=from_account_id, body=from_account)
    batch.replace_item(item=to_account_id, body=to_account)
    
    results = container.execute_transactional_batch(batch=batch)
    
    # Check results
    for result in results:
        if result["statusCode"] >= 400:
            raise Exception(f"Transaction failed: {result}")
    
    return results
```

### Batch Operations

```python
from azure.cosmos import TransactionalBatch

def batch_create_items(container, items: list[dict], partition_key: str):
    """Create multiple items in a single batch (max 100 operations)."""
    batch = TransactionalBatch(partition_key=partition_key)
    
    for item in items[:100]:  # Max 100 operations per batch
        batch.create_item(body=item)
    
    results = container.execute_transactional_batch(batch=batch)
    
    successful = sum(1 for r in results if r["statusCode"] < 400)
    print(f"Created {successful}/{len(items)} items")
    
    return results

# Usage
items = [
    {"id": f"item-{i}", "pk": "batch-test", "value": i}
    for i in range(50)
]
batch_create_items(container, items, partition_key="batch-test")
```

## Change Feed

Process changes to documents in order:

```python
def process_change_feed(container, partition_key: str | None = None):
    """Process change feed for a container."""
    
    # Start from beginning
    change_feed = container.query_items_change_feed(
        partition_key=partition_key,  # None for all partitions
        start_time="Beginning"
    )
    
    for change in change_feed:
        print(f"Changed document: {change['id']}")
        # Process the change
        process_document(change)
    
    # Get continuation token for next run
    continuation = change_feed.continuation_token
    return continuation

def process_change_feed_incremental(container, continuation_token: str):
    """Resume change feed from continuation token."""
    change_feed = container.query_items_change_feed(
        continuation_token=continuation_token
    )
    
    for change in change_feed:
        process_document(change)
    
    return change_feed.continuation_token
```

## Query Best Practices

| Practice | Recommendation |
|----------|----------------|
| Always use partition key | Avoid cross-partition queries when possible |
| Use parameters | Enable query plan caching, prevent injection |
| Project only needed fields | Reduces RU cost and bandwidth |
| Use point reads | For single document by ID, use `read_item` not query |
| Index strategy | Create composite indexes for ORDER BY on multiple fields |
| Pagination | Use continuation tokens for large result sets |
| Aggregations | Prefer single-partition aggregations |
| Avoid OFFSET | Use continuation tokens instead for large datasets |

## RU Estimation

```python
def estimate_query_cost(container, query: str, partition_key: str, parameters: list | None = None):
    """Execute query and return RU cost."""
    
    query_iterable = container.query_items(
        query=query,
        parameters=parameters or [],
        partition_key=partition_key,
        populate_query_metrics=True
    )
    
    items = list(query_iterable)
    
    # Get request charge from response headers
    request_charge = query_iterable.get_response_headers().get("x-ms-request-charge", 0)
    
    print(f"Query returned {len(items)} items")
    print(f"Total RU cost: {request_charge}")
    print(f"RU per item: {float(request_charge) / len(items) if items else 0:.2f}")
    
    return items, float(request_charge)
```

````


### `scripts/setup_cosmos_container.py`

```
#!/usr/bin/env python3
"""
Cosmos DB Container Setup CLI Tool

Create and configure Azure Cosmos DB containers with proper partitioning,
throughput settings, and indexing policies.

Usage:
    python setup_cosmos_container.py --database mydb --container orders --partition-key /customer_id
    python setup_cosmos_container.py --database mydb --container events --partition-key /device_id /day --throughput 1000
    python setup_cosmos_container.py --database mydb --container data --partition-key /pk --serverless

Environment Variables:
    COSMOS_ENDPOINT  - Cosmos DB account endpoint URL
    COSMOS_KEY       - Cosmos DB account key (optional if using DefaultAzureCredential)
"""

import argparse
import json
import os
import sys
from typing import Any

from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient, PartitionKey
from azure.cosmos.exceptions import CosmosHttpResponseError


def get_cosmos_client() -> CosmosClient:
    """Create Cosmos DB client from environment variables."""
    endpoint = os.environ.get("COSMOS_ENDPOINT")
    if not endpoint:
        raise ValueError("COSMOS_ENDPOINT environment variable required")
    
    # Try key auth first, fall back to DefaultAzureCredential
    key = os.environ.get("COSMOS_KEY")
    if key:
        return CosmosClient(url=endpoint, credential=key)
    else:
        credential = DefaultAzureCredential()
        return CosmosClient(url=endpoint, credential=credential)


def create_indexing_policy(
    include_paths: list[str] | None = None,
    exclude_paths: list[str] | None = None,
    composite_indexes: list[list[dict]] | None = None
) -> dict[str, Any]:
    """Build an indexing policy."""
    policy = {
        "indexingMode": "consistent",
        "automatic": True,
        "includedPaths": [],
        "excludedPaths": []
    }
    
    # Include paths (default: all)
    if include_paths:
        policy["includedPaths"] = [{"path": p} for p in include_paths]
    else:
        policy["includedPaths"] = [{"path": "/*"}]
    
    # Exclude paths
    if exclude_paths:
        policy["excludedPaths"] = [{"path": p} for p in exclude_paths]
    
    # Always exclude _etag
    policy["excludedPaths"].append({"path": "/_etag/?")})
    
    # Composite indexes for ORDER BY on multiple fields
    if composite_indexes:
        policy["compositeIndexes"] = composite_indexes
    
    return policy


def create_container(
    client: CosmosClient,
    database_id: str,
    container_id: str,
    partition_key_paths: list[str],
    throughput: int | None = None,
    ttl: int | None = None,
    indexing_policy: dict | None = None
) -> dict[str, Any]:
    """Create or update a Cosmos DB container."""
    
    # Get or create database
    try:
        database = client.create_database_if_not_exists(id=database_id)
        print(f"Database: {database_id}")
    except CosmosHttpResponseError as e:
        print(f"Error creating database: {e.message}")
        raise
    
    # Build partition key
    if len(partition_key_paths) == 1:
        partition_key = PartitionKey(path=partition_key_paths[0])
    else:
        # Hierarchical partition key
        partition_key = PartitionKey(path=partition_key_paths)
    
    # Container properties
    container_props = {
        "id": container_id,
        "partition_key": partition_key
    }
    
    # Add TTL if specified
    if ttl is not None:
        container_props["default_time_to_live"] = ttl
    
    # Add indexing policy if specified
    if indexing_policy:
        container_props["indexing_policy"] = indexing_policy
    
    # Create container
    try:
        if throughput:
            container = database.create_container_if_not_exists(
                **container_props,
                offer_throughput=throughput
            )
        else:
            container = database.create_container_if_not_exists(**container_props)
        
        print(f"Container: {container_id}")
        print(f"Partition key: {partition_key_paths}")
        
    except CosmosHttpResponseError as e:
        if e.status_code == 409:
            print(f"Container {container_id} already exists")
            container = database.get_container_client(container_id)
        else:
            print(f"Error creating container: {e.message}")
            raise
    
    # Get container properties
    properties = container.read()
    
    return {
        "database": database_id,
        "container": container_id,
        "partition_key": partition_key_paths,
        "self_link": properties.get("_self"),
        "resource_id": properties.get("_rid")
    }


def show_container_info(client: CosmosClient, database_id: str, container_id: str):
    """Display detailed container information."""
    database = client.get_database_client(database_id)
    container = database.get_container_client(container_id)
    
    properties = container.read()
    
    print("\n=== Container Information ===")
    print(f"Database: {database_id}")
    print(f"Container: {container_id}")
    print(f"Partition Key: {properties.get('partitionKey', {}).get('paths', [])}")
    
    # TTL
    ttl = properties.get("defaultTtl")
    if ttl == -1:
        print("TTL: Enabled (per-item)")
    elif ttl:
        print(f"TTL: {ttl} seconds")
    else:
        print("TTL: Disabled")
    
    # Indexing policy
    index_policy = properties.get("indexingPolicy", {})
    print(f"Indexing Mode: {index_policy.get('indexingMode', 'consistent')}")
    
    # Throughput
    try:
        offer = container.read_offer()
        print(f"Throughput: {offer.offer_throughput} RU/s")
        if offer.properties.get("content", {}).get("offerAutopilotSettings"):
            max_throughput = offer.properties["content"]["offerAutopilotSettings"]["maxThroughput"]
            print(f"Autoscale Max: {max_throughput} RU/s")
    except Exception:
        print("Throughput: Serverless or database-level")
    
    # Item count (approximate)
    try:
        query = "SELECT VALUE COUNT(1) FROM c"
        count = list(container.query_items(query=query, enable_cross_partition_query=True))[0]
        print(f"Item Count: ~{count}")
    except Exception:
        print("Item Count: Unable to retrieve")


def main():
    parser = argparse.ArgumentParser(
        description="Create and configure Cosmos DB containers",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__
    )
    
    parser.add_argument(
        "--database", "-d",
        required=True,
        help="Database ID"
    )
    parser.add_argument(
        "--container", "-c",
        required=True,
        help="Container ID"
    )
    parser.add_argument(
        "--partition-key", "-pk",
        nargs="+",
        required=True,
        help="Partition key path(s). Use multiple for hierarchical keys."
    )
    parser.add_argument(
        "--throughput", "-t",
        type=int,
        help="Provisioned throughput in RU/s (omit for serverless)"
    )
    parser.add_argument(
        "--serverless",
        action="store_true",
        help="Use serverless mode (no throughput provisioning)"
    )
    parser.add_argument(
        "--ttl",
        type=int,
        help="Default TTL in seconds (-1 for per-item TTL)"
    )
    parser.add_argument(
        "--exclude-paths",
        nargs="+",
        help="Paths to exclude from indexing (e.g., /large_field/*)"
    )
    parser.add_argument(
        "--composite-index",
        action="append",
        nargs="+",
        metavar="PATH",
        help="Composite index paths (can specify multiple times). Format: --composite-index /field1 /field2"
    )
    parser.add_argument(
        "--info",
        action="store_true",
        help="Show container information instead of creating"
    )
    parser.add_argument(
        "--output", "-o",
        choices=["json", "text"],
        default="text",
        help="Output format (default: text)"
    )
    
    args = parser.parse_args()
    
    # Validate arguments
    if args.throughput and args.serverless:
        print("Error: Cannot specify both --throughput and --serverless")
        sys.exit(1)
    
    # Ensure partition key paths start with /
    partition_keys = []
    for pk in args.partition_key:
        if not pk.startswith("/"):
            pk = f"/{pk}"
        partition_keys.append(pk)
    
    try:
        client = get_cosmos_client()
    except ValueError as e:
        print(f"Error: {e}")
        sys.exit(1)
    
    # Show info mode
    if args.info:
        try:
            show_container_info(client, args.database, args.container)
        except CosmosHttpResponseError as e:
            print(f"Error: {e.message}")
            sys.exit(1)
        return
    
    # Build indexing policy
    indexing_policy = None
    if args.exclude_paths or args.composite_index:
        composite_indexes = None
        if args.composite_index:
            composite_indexes = [
                [{"path": p, "order": "ascending"} for p in index_paths]
                for index_paths in args.composite_index
            ]
        
        indexing_policy = create_indexing_policy(
            exclude_paths=args.exclude_paths,
            composite_indexes=composite_indexes
        )
    
    # Create container
    try:
        result = create_container(
            client=client,
            database_id=args.database,
            container_id=args.container,
            partition_key_paths=partition_keys,
            throughput=args.throughput if not args.serverless else None,
            ttl=args.ttl,
            indexing_policy=indexing_policy
        )
    except CosmosHttpResponseError as e:
        print(f"Error: {e.message}")
        sys.exit(1)
    
    # Output result
    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print("\n=== Container Created ===")
        print(f"Database: {result['database']}")
        print(f"Container: {result['container']}")
        print(f"Partition Key: {result['partition_key']}")
        if args.throughput:
            print(f"Throughput: {args.throughput} RU/s")
        else:
            print("Throughput: Serverless")
        if args.ttl:
            print(f"TTL: {args.ttl} seconds")
        
        print("\nContainer ready for use!")


if __name__ == "__main__":
    main()

```