Develop AI Solutions with Azure Cosmos DB for NoSQL

1. Resource Hierarchy

Memory aid: Library → Floor → Bookshelf → Book

1. Account — top-level resource. Unique DNS endpoint: https://myaccount.documents.azure.com:443/. Holds up to 500 databases.
2. Database — logical namespace. Groups related containers. Can share throughput across containers (shared throughput mode).
3. Container — unit of scalability. Requires a partition key. Schema-agnostic — items in the same container can have different shapes.
4. Item — individual JSON document. Must have id + partition key value to be uniquely addressable via point read.

2. Partition Key Design — The Most Important Decision

1. High cardinality — many distinct values spread data across partitions evenly. Good: userId, tenantId, productId. Bad: status (only true/false), isActive (boolean).
2. Query-aligned — if you always query by userId, make it the partition key. Same-partition queries cost fewer RUs and are faster.
3. Immutable — you cannot change a container's partition key. Wrong choice = create new container + migrate all data.

3. Request Units (RUs) — Cost Model

1. Point read 1 KB item ≈ 1 RU — cheapest operation. Requires both id AND partition key.
2. Write 1 KB item ≈ 5–10 RUs — always more expensive than reads.
3. Simple query (with partition key) ≈ 3–5 RUs
4. Cross-partition query (scan) ≈ 100+ RUs — fan-out to all partitions. Avoid in hot paths.

Monitor cost via the x-ms-request-charge response header or response.request_charge in SDK.

4. Throughput Modes

1. Manual throughput — fixed RU/s you specify. Min 400 RU/s for dedicated container. Predictable cost. Use for steady, known workloads.
2. Autoscale throughput — scales 10%–100% of your set maximum automatically. Min 1,000 RU/s. No over-provisioning. Best for AI workloads with bursty or unpredictable traffic.

5. Item System Properties

1. _etag — optimistic concurrency token. Use with if_match to prevent lost updates from race conditions.
2. _ts — Unix timestamp of last modification. Use for change feed or auditing.
3. _rid, _self — internal resource identifiers (read-only, auto-generated).

1. Client Initialization — Singleton Pattern

Create CosmosClient ONCE at application startup and reuse it. Recreating per request wastes connections and adds latency (TCP handshake overhead).

from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential

client = CosmosClient(
    "https://myaccount.documents.azure.com:443/",
    credential=DefaultAzureCredential()   # Managed identity in prod, az CLI locally
)
db = client.get_database_client("aidata")
container = db.get_container_client("products")

2. Authentication Options

1. Account keys — simple, full access. Use for dev/test only. Avoid in production (long-lived, no audit trail).
2. Entra ID (RBAC) — production standard. Assign Cosmos DB Built-in Data Contributor (read/write) or Cosmos DB Built-in Data Reader (read-only). Works with managed identities.

3. CRUD Operations

# Create — fails if id already exists
container.create_item(body={"id": "product-123", "categoryId": "electronics", "name": "Speaker"})

# Upsert — insert or replace (no conflict error)
container.upsert_item(body={...})

# Point Read — cheapest: ~1 RU for 1 KB (REQUIRES both id AND partition key)
item = container.read_item(item="product-123", partition_key="electronics")

# Optimistic concurrency — prevent lost updates
container.replace_item(item=item["id"], body=item, if_match=item["_etag"])

# Delete
container.delete_item(item="product-123", partition_key="electronics")

4. Idempotent Container Creation

from azure.cosmos import PartitionKey, ThroughputProperties

db = client.create_database_if_not_exists(id="aidata")
container = db.create_container_if_not_exists(
    id="products",
    partition_key=PartitionKey(path="/categoryId"),
    offer_throughput=ThroughputProperties(auto_scale_max_throughput=4000)
)

1. SQL Syntax (JSON-aware)

# Basic filter + projection (c = container alias)
SELECT c.id, c.name, c.price
FROM c
WHERE c.categoryId = "electronics"
  AND c.price < 100

# Array contains
SELECT c.id FROM c WHERE ARRAY_CONTAINS(c.features, "wifi")

2. Parameterized Queries via SDK

query = "SELECT * FROM c WHERE c.categoryId = @cat AND c.price < @price"
params = [{"name": "@cat", "value": "electronics"}, {"name": "@price", "value": 100}]

items = list(container.query_items(
    query=query,
    parameters=params,
    partition_key="electronics"   # Stays in one partition = cheap
))

3. Cross-Partition Queries — Avoid in Hot Paths

Queries without the partition key in WHERE clause fan out to ALL partitions. They cost dramatically more RUs and have higher latency. If you frequently need cross-partition queries, reconsider your partition key choice.

Configure Vector Index on Container

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

client = CosmosClient(
    "https://myacct.documents.azure.com:443/",
    credential=DefaultAzureCredential()
)
db = client.get_database_client("ai-db")

# Create container with vector policy
container = db.create_container_if_not_exists(
    id="documents",
    partition_key=PartitionKey("/category"),
    vector_embedding_policy={
        "vectorEmbeddings": [{
            "path": "/embedding",
            "dataType": "float32",
            "dimensions": 1536,
            "distanceFunction": "cosine"
        }]
    },
    indexing_policy={
        "vectorIndexes": [{
            "path": "/embedding",
            "type": "diskANN"
        }]
    }
)

VectorDistance() Query

def search_documents(query_text, top_k=5):
    q_emb = generate_embedding(query_text)
    results = list(container.query_items(
        query="""SELECT TOP @k c.title, c.content,
            VectorDistance(c.embedding, @qvec) AS score
            FROM c
            ORDER BY VectorDistance(c.embedding, @qvec)""",
        parameters=[
            {"name": "@k", "value": top_k},
            {"name": "@qvec", "value": q_emb}
        ],
        enable_cross_partition_query=True
    ))
    return results

Process New Documents Automatically

from azure.cosmos.aio import CosmosClient
import asyncio

async def process_changes(changes, context):
    for doc in changes:
        # Auto-embed every new document inserted to container
        text = doc.get("content", "")
        embedding = await generate_embedding_async(text)
        await container.patch_item(
            item=doc["id"],
            partition_key=doc["category"],
            patch_operations=[{
                "op": "add",
                "path": "/embedding",
                "value": embedding
            }]
        )

# Start change feed processor
processor = container.create_change_feed_processor(
    name="embedding-processor",
    handler=process_changes,
    lease_container=lease_container
)
await processor.start()