In this post we are going to cover data modeling in Azure Cosmos DB. These are 7 real Cosmos DB interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: you are moving an e-commerce schema with customers, orders and products from SQL to Cosmos DB, and you model it.
Data modeling in Cosmos DB means deciding what goes into one JSON item, what goes into its own item, and which items share a container and a partition key. Interviewers ask about it because most people arrive from SQL Server, and the habits that work there, normalize first and join later, make a Cosmos DB app slow and expensive. If you can explain embed versus reference and keep copies in sync, you show that you've built one.
Part 4 of the Real Time Azure Cosmos DB Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A teammate coming from SQL Server asks how to model data in Cosmos DB. Where do you start?
- Question 2: Someone asks when to embed related data and when to reference it. How do you decide?
- Question 3: An order item list keeps growing without limit. What goes wrong, and how do you model it?
- Question 4: The same customer name is copied into many documents. How do you keep denormalized data in sync?
- Question 5: A teammate wants to store different entity types in one container. When does that make sense?
- Question 6: Someone asks how to model many-to-many relationships. What do you suggest?
- Scenario 7: You are moving an e-commerce schema with customers, orders and products from SQL to Cosmos DB. How do you model it?
- The full script
The short answer
Model from the access patterns, not the tables: embed what is read together and bounded, reference what grows or is shared, keep items under the 2 MB limit, and sync denormalized copies from the change feed. Several types can share a container when they share the partition key.
Let's set up the project
We run the official Azure Cosmos DB Linux emulator locally, and small Python scripts that use the azure-cosmos client, so every output is real. Features the emulator doesn't support are shown as code. common.py creates the client from the COSMOS_ENDPOINT and COSMOS_KEY environment variables, and the shop database as db.
import os
from azure.cosmos import CosmosClient
client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")Question 1: A teammate coming from SQL Server asks how to model data in Cosmos DB. Where do you start?
Start from the access patterns, not from the tables. In SQL Server we normalize first and write the queries later. In Cosmos DB we list what the app reads and writes, and how often, and shape the items for that.
Let's put the two side by side:
| SQL Server | Cosmos DB (API for NoSQL) | |
|---|---|---|
| Start from | tables and normal forms | the queries the app runs |
| Joins | across tables, at query time | inside one item only |
| Transactions | across any rows | inside one logical partition |
| Schema | fixed per table | per item, flexible |
The reason is how the engine works. A join only works inside one item, never across items or containers. And a transaction only spans one logical partition. So data we need in one call should live in one item, or at least in one logical partition, under one partition key.
My order of work is:
- List the access patterns - reads and writes, with their volume
- Pick the partition key - from the busiest patterns
- Embed - what is read together and stays bounded
- Reference - what grows without limit or is shared
- Plan the copies - how denormalized data stays in sync
Copying the SQL schema one table to one container. Then every page needs several reads, and the bill shows it.
Question 2: Someone asks when to embed related data and when to reference it. How do you decide?
Embed when the data is read together, belongs to the parent and stays bounded, like the lines of an order. Reference when it grows without limit, is shared by many parents, or changes on its own, like a product.
| Embed when | Reference when | |
|---|---|---|
| Read | together, on the same page | separately |
| Owner | belongs to the parent | shared by many parents |
| Size | bounded | grows without limit |
| Changes | with the parent | on its own |
Let's build both. load.py writes three products and two orders. Order o1 embeds each line with the product name and price. Order o2 keeps only the product key: the id and the category, which is the partition key of products. Then reads.py shows each order page and counts the calls. We run python load.py and python reads.py.
from azure.cosmos import PartitionKey
from common import db
orders = db.create_container_if_not_exists(
"orders", partition_key=PartitionKey(path="/customerId"))
products = db.create_container_if_not_exists(
"products", partition_key=PartitionKey(path="/category"))
catalog = [("p1", "kitchen", "Kettle", 29.5),
("p2", "kitchen", "Mug", 7.5),
("p3", "pantry", "Tea", 4.25)]
for pid, cat, name, price in catalog:
products.upsert_item({"id": pid, "category": cat,
"name": name, "price": price})
qty = {"p1": 1, "p2": 2, "p3": 3}
# A: embedded - each line carries what the order page shows
orders.upsert_item({"id": "o1", "customerId": "c1", "lines": [
{"productId": p, "name": n, "price": pr, "qty": qty[p]}
for p, _, n, pr in catalog]})
# B: referenced - each line keeps only the product key
orders.upsert_item({"id": "o2", "customerId": "c1", "lines": [
{"productId": p, "category": c, "qty": qty[p]}
for p, c, _, _ in catalog]})
print("products: 3, orders: o1 embedded, o2 referenced")from common import db
orders = db.get_container_client("orders")
products = db.get_container_client("products")
calls = 1 # A: embedded
order = orders.read_item("o1", partition_key="c1")
lines = [(l["name"], l["qty"]) for l in order["lines"]]
print("embedded :", lines, "| calls:", calls)
calls = 1 # B: referenced
order = orders.read_item("o2", partition_key="c1")
lines = []
for l in order["lines"]:
p = products.read_item(l["productId"],
partition_key=l["category"])
calls += 1
lines.append((p["name"], l["qty"]))
print("referenced:", lines, "| calls:", calls)$ python load.py
products: 3, orders: o1 embedded, o2 referenced
$ python reads.py
embedded : [('Kettle', 1), ('Mug', 2), ('Tea', 3)] | calls: 1
referenced: [('Kettle', 1), ('Mug', 2), ('Tea', 3)] | calls: 4Both pages show the same lines, [('Kettle', 1), ('Mug', 2), ('Tea', 3)]. The embedded order needs calls: 1, one point read. The referenced one needs a read for the order plus one per product: calls: 4 for the same page. read_items can cut that to two, never to one.
Embedding the price is also the right business rule. The order must keep the price the customer paid, even when the catalog price changes later.
I often mix the two: embed a small snapshot, like name and price, and keep the product id as a reference for the full details.
Question 3: An order item list keeps growing without limit. What goes wrong, and how do you model it?
Two things go wrong. Every append rewrites the whole item, so each write gets bigger and costs more. And at 2 MB the item hits a hard limit, and the writes start failing.
| Limit (Azure) | Value |
|---|---|
| item size | 2 MB |
| logical partition | 20 GB |
Let's grow one. grow.py appends 17,500 lines per round to order o9 and saves it again. Then lines_as_items.py models the same order the right way. We run python grow.py and python lines_as_items.py.
import json
from azure.cosmos.exceptions import CosmosHttpResponseError
from common import db
orders = db.get_container_client("orders")
order = {"id": "o9", "customerId": "c9", "lines": []}
for rnd in [1, 2, 3]:
start = len(order["lines"])
order["lines"] += [{"sku": f"SKU-{start + i:06}", "qty": 1,
"price": 9.5} for i in range(17500)]
size = len(json.dumps(order, separators=(",", ":")))
print(f"round {rnd}: {len(order['lines'])} lines, {size} bytes")
try:
orders.upsert_item(order) # rewrites the whole item
print(" saved")
except CosmosHttpResponseError as e:
print(" ", type(e).__name__, e.status_code)
print(" ", str(e).split("\n")[0])import json
from azure.cosmos import PartitionKey
from common import db
lines = db.create_container_if_not_exists(
"order_lines", partition_key=PartitionKey(path="/orderId"))
lines.upsert_item({"id": "o9", "orderId": "o9", "type": "header",
"customerId": "c9", "lineCount": 500})
for i in range(500): # one small item per line
lines.upsert_item({"id": f"o9-{i:04}", "orderId": "o9",
"type": "line", "sku": f"SKU-{i:06}",
"qty": 1, "price": 9.5})
q = "SELECT VALUE COUNT(1) FROM c WHERE c.type = 'line'"
print("line items in partition o9:",
list(lines.query_items(q, partition_key="o9"))[0])
pager = lines.query_items("SELECT c.sku FROM c WHERE c.type = 'line'",
partition_key="o9", max_item_count=100)
page = list(next(pager.by_page()))
print("first page:", len(page), "lines, from", page[0]["sku"])
one = lines.read_item("o9-0000", partition_key="o9")
body = {k: v for k, v in one.items() if not k.startswith("_")}
size = len(json.dumps(body, separators=(",", ":")))
print("one line item:", size, "bytes")$ python grow.py
round 1: 17500 lines, 717539 bytes
saved
round 2: 35000 lines, 1435039 bytes
saved
round 3: 52500 lines, 2152539 bytes
CosmosHttpResponseError 413
(RequestEntityTooLarge) Entity with the specified id is larger than the maximum entity size (2 MiB).
Entity size: 2152539 bytes.
$ python lines_as_items.py
line items in partition o9: 500
first page: 100 lines, from SKU-000000
one line item: 84 bytesRound 1 is 717539 bytes and round 2 is 1435039 bytes. Both are saved. Round 3 is 2152539 bytes, and the write fails: CosmosHttpResponseError 413, (RequestEntityTooLarge), larger than the maximum entity size of 2 MiB.
The fix: each line becomes its own small item in order_lines, keyed by /orderId, next to a header item. 500 lines land in one logical partition, o9. One query counts them, and paging reads 100 lines at a time, starting from SKU-000000. Each line item is only 84 bytes.
That's per logical partition. For data that never stops, like events, bucket the key, for example customer and month, or use a hierarchical partition key.
Question 4: The same customer name is copied into many documents. How do you keep denormalized data in sync?
Treat the customer item as the source of truth, and update the copies from its change feed. Every created or updated customer shows up in the feed, in its latest version, so a small fixer job can patch every copy.
- customers - c2 renamed
- Change feed - from a saved token
- Fixer - query stale copies
- reviews - patch customerName
Let's run it. Each review stores the customer name, so a product page needs no second read. seed.py writes two customers and four reviews, and saves the change feed position in a file. rename.py renames customer c2. Then fixer.py reads the feed from the saved token, and check.py lists the reviews. Last, we run the fixer once more. We run python seed.py, python rename.py, python fixer.py, python check.py and python fixer.py.
from azure.cosmos import PartitionKey
from common import db
customers = db.create_container_if_not_exists(
"customers", partition_key=PartitionKey(path="/customerId"))
reviews = db.create_container_if_not_exists(
"reviews", partition_key=PartitionKey(path="/productId"))
for cid, name in [("c2", "Ana Lopez"), ("c3", "Raj Patel")]:
customers.upsert_item({"id": cid, "customerId": cid,
"name": name})
for rid, pid, cid, name in [("r1", "p1", "c2", "Ana Lopez"),
("r2", "p2", "c2", "Ana Lopez"),
("r3", "p3", "c2", "Ana Lopez"),
("r4", "p1", "c3", "Raj Patel")]:
reviews.upsert_item({"id": rid, "productId": pid, "stars": 5,
"customerId": cid, "customerName": name})
list(customers.query_items_change_feed(start_time="Beginning"))
token = customers.client_connection.last_response_headers["etag"]
open("customers.token", "w").write(token) # fixer starts here
print("2 customers, 4 reviews; change feed position saved")from common import db
customers = db.get_container_client("customers")
customers.patch_item("c2", partition_key="c2", patch_operations=[
{"op": "set", "path": "/name", "value": "Ana Lopez-Garcia"}])
print("customer c2 renamed")from common import db
customers = db.get_container_client("customers")
reviews = db.get_container_client("reviews")
feed = customers.query_items_change_feed(
continuation=open("customers.token").read())
changes = list(feed) # customers changed since
token = customers.client_connection.last_response_headers["etag"]
Q = ("SELECT c.id, c.productId FROM c WHERE c.customerId = @id"
" AND c.customerName != @name")
for cust in changes:
args = [{"name": "@id", "value": cust["customerId"]},
{"name": "@name", "value": cust["name"]}]
stale = list(reviews.query_items(
Q, parameters=args, enable_cross_partition_query=True))
for r in stale: # copies to fix
reviews.patch_item(r["id"], partition_key=r["productId"],
patch_operations=[{"op": "set", "path": "/customerName",
"value": cust["name"]}])
print("change:", cust["id"], cust["name"], "| reviews patched:",
sorted(r["id"] for r in stale))
print("changes read:", len(changes))
open("customers.token", "w").write(token) # save after the workfrom common import db
reviews = db.get_container_client("reviews")
rows = reviews.query_items("SELECT c.id, c.customerName FROM c",
enable_cross_partition_query=True)
for r in sorted(rows, key=lambda r: r["id"]):
print(r["id"], r["customerName"])$ python seed.py
2 customers, 4 reviews; change feed position saved
$ python rename.py
customer c2 renamed
$ python fixer.py
change: c2 Ana Lopez-Garcia | reviews patched: ['r1', 'r2', 'r3']
changes read: 1
$ python check.py
r1 Ana Lopez-Garcia
r2 Ana Lopez-Garcia
r3 Ana Lopez-Garcia
r4 Raj Patel
$ python fixer.py
changes read: 0The fixer finds changes read: 1: c2 Ana Lopez-Garcia. It queries the reviews that still carry the old name and patches three of them, ['r1', 'r2', 'r3']. The check shows all three as Ana Lopez-Garcia, and r4 Raj Patel is untouched.
The fixer saves the new token only after the work, so a crash means a retry, not a gap. The second run reads changes read: 0. The Python SDK has no change feed processor, so a script uses this pull model with a saved continuation token.
The copies are in sync a little later, not instantly. So copy fields that change rarely, like a name, never fast-moving ones, like stock.
Question 5: A teammate wants to store different entity types in one container. When does that make sense?
When the types share a partition key and are read together. Containers are schema-agnostic, so a customer, a cart and orders can live side by side, and a type field tells them apart.
Let's put them together. one_container.py writes customer c5, a cart and two orders into customer_data, all with partition key c5. Then customer c6 and a first order go in as one transactional batch: same partition, so all or nothing. Then it counts by type, and loads everything for c5. We run python one_container.py.
from azure.cosmos import PartitionKey
from common import db
data = db.create_container_if_not_exists(
"customer_data", partition_key=PartitionKey(path="/customerId"))
for d in [{"id": "c5", "type": "customer", "name": "Lena Kim"},
{"id": "cart-c5", "type": "cart", "items": 2},
{"id": "o51", "type": "order", "total": 36.5},
{"id": "o52", "type": "order", "total": 12.25}]:
data.upsert_item(dict(d, customerId="c5"))
c6 = {"id": "c6", "customerId": "c6", "type": "customer",
"name": "Omar Haddad"}
o61 = {"id": "o61", "customerId": "c6", "type": "order",
"total": 8.75}
data.execute_item_batch([("upsert", (c6,)), ("upsert", (o61,))],
partition_key="c6") # all or nothing
Q = "SELECT VALUE COUNT(1) FROM c WHERE c.type = @t"
for t in ["customer", "cart", "order"]:
n = list(data.query_items(Q, enable_cross_partition_query=True,
parameters=[{"name": "@t", "value": t}]))[0]
print(f"type = {t:9} -> {n}")
rows = data.query_items("SELECT c.type, c.id FROM c ORDER BY c.type",
partition_key="c5") # one partition
print("c5:", [(r["type"], r["id"]) for r in rows])$ python one_container.py
type = customer -> 2
type = cart -> 1
type = order -> 3
c5: [('cart', 'cart-c5'), ('customer', 'c5'), ('order', 'o51'), ('order', 'o52')]Queries stay simple. Filtering on the type gives customer -> 2, cart -> 1 and order -> 3. And one single-partition query returns everything for c5, sorted by type: the cart, the customer, and orders o51 and o52.
That's the win: one call loads the whole customer page, and transactions work across the types.
- Same partition key - read together in one call
- Transactions - batch across the types
- Keep apart - different key, index policy or throughput
Add the type field to every item. It's how queries and readers tell the types apart.
Question 6: Someone asks how to model many-to-many relationships. What do you suggest?
Keep a list of references on both sides. A product lists the collections it belongs to, and a collection lists its products, each with its id and partition key, so every lookup is a direct read.
Let's link them. The kettle is in gifts and home, and the mug and the tea are in gifts. link.py writes both sides: a patch on each product that sets collectionIds, and one item per collection with its products. Then browse.py reads both sides and runs one query. We run python link.py and python browse.py.
from azure.cosmos import PartitionKey
from common import db
products = db.get_container_client("products")
collections = db.create_container_if_not_exists(
"collections", partition_key=PartitionKey(path="/id"))
# side 1: each product lists its collections
for pid, cat, ids in [("p1", "kitchen", ["gifts", "home"]),
("p2", "kitchen", ["gifts"]),
("p3", "pantry", ["gifts"])]:
products.patch_item(pid, partition_key=cat, patch_operations=[
{"op": "set", "path": "/collectionIds", "value": ids}])
# side 2: each collection lists its products (id + partition key)
collections.upsert_item({"id": "gifts", "products": [
["p1", "kitchen"], ["p2", "kitchen"], ["p3", "pantry"]]})
collections.upsert_item({"id": "home", "products": [
["p1", "kitchen"]]})
print("linked: 3 products, 2 collections")from common import db
products = db.get_container_client("products")
collections = db.get_container_client("collections")
gifts = collections.read_item("gifts", partition_key="gifts")
keys = [tuple(k) for k in gifts["products"]]
found = products.read_items(keys) # one call, three items
print("collection gifts ->", sorted(p["name"] for p in found))
kettle = products.read_item("p1", partition_key="kitchen")
print("product p1 ->", kettle["collectionIds"])
Q = ("SELECT VALUE c.name FROM c"
" WHERE ARRAY_CONTAINS(c.collectionIds, @g)")
rows = products.query_items(Q, parameters=[{"name": "@g",
"value": "home"}],
enable_cross_partition_query=True)
print("query, products in home ->", sorted(rows))$ python link.py
linked: 3 products, 2 collections
$ python browse.py
collection gifts -> ['Kettle', 'Mug', 'Tea']
product p1 -> ['gifts', 'home']
query, products in home -> ['Kettle']To show the gifts page, we do one point read of the collection, then one read_items call for its three products: ['Kettle', 'Mug', 'Tea']. The kettle's page already holds its collections in its own item: ['gifts', 'home'].
We can also query one side with ARRAY_CONTAINS, here the products in home: ['Kettle']. But without the partition key in the filter, that query has to look in every partition.
The two sides live in different containers, so one link is two writes, not one transaction. Make one side the source of truth, and fix the other from the change feed.
Scenario 7: You are moving an e-commerce schema with customers, orders and products from SQL to Cosmos DB. How do you model it?
I start with the access patterns, and only then draw containers. The busy ones are:
- Customer profile - by customer id
- Order history - a customer's orders, newest first
- One order with its items - the order page
- Product page - by product id
- Browse a category - sorted by price
- Sell an item - stock minus one
That gives two containers. customers, keyed by /customerId, holds the customer and their orders, with the order lines embedded. products, keyed by /category, holds the catalog. Four SQL tables, Customers, Orders, OrderLines and Products, become two containers.
| Access pattern | Container | Partition key | Operation |
|---|---|---|---|
| customer profile | customers | customerId | point read |
| order history | customers | customerId | query, one partition |
| one order with its items | customers | customerId | point read |
| product page | products | category | point read |
| browse a category | products | category | query, one partition |
| sell an item | products | category | patch incr -1, if stock > 0 |
Let's run each pattern. model.py loads one customer, two orders and three products into the ecommerce database, and app.py runs the six patterns. We run python model.py and python app.py.
from azure.cosmos import PartitionKey
from common import client
shop = client.create_database_if_not_exists("ecommerce")
customers = shop.create_container_if_not_exists( # customer + orders
"customers", partition_key=PartitionKey(path="/customerId"))
products = shop.create_container_if_not_exists(
"products", partition_key=PartitionKey(path="/category"))
for pid, cat, name, price, stock in [
("p1", "kitchen", "Kettle", 29.5, 12),
("p2", "kitchen", "Mug", 7.5, 40),
("p3", "pantry", "Tea", 4.25, 100)]:
products.upsert_item({"id": pid, "category": cat, "name": name,
"price": price, "stock": stock})
customers.upsert_item({"id": "c1", "customerId": "c1",
"type": "customer", "name": "Ana Lopez"})
for oid, day, items in [
("o1", "2026-09-02", [["p3", "Tea", 4.25, 2]]),
("o2", "2026-10-14", [["p1", "Kettle", 29.5, 1],
["p2", "Mug", 7.5, 2]])]:
customers.upsert_item({"id": oid, "customerId": "c1",
"type": "order", "date": day, "items": [dict(zip(
["productId", "name", "price", "qty"], i)) for i in items]})
print("ecommerce: 1 customer, 2 orders, 3 products")from common import client
shop = client.get_database_client("ecommerce")
customers = shop.get_container_client("customers")
products = shop.get_container_client("products")
me = customers.read_item("c1", partition_key="c1")
print("1 profile point read :", me["name"])
Q = ("SELECT VALUE c.id FROM c WHERE c.type = 'order'"
" ORDER BY c.date DESC")
mine = list(customers.query_items(Q, partition_key="c1"))
print("2 order history 1 partition :", mine)
o2 = customers.read_item("o2", partition_key="c1")
print("3 order + items point read :",
[(i["name"], i["qty"]) for i in o2["items"]])
p = products.read_item("p1", partition_key="kitchen")
print("4 product page point read :", p["name"], p["price"])
Q = "SELECT VALUE c.name FROM c ORDER BY c.price"
print("5 browse kitchen 1 partition :",
list(products.query_items(Q, partition_key="kitchen")))
after = products.patch_item("p1", partition_key="kitchen",
patch_operations=[{"op": "incr", "path": "/stock", "value": -1}],
filter_predicate="FROM c WHERE c.stock > 0") # fails if sold out
print("6 sell one patch : stock", after["stock"])$ python model.py
ecommerce: 1 customer, 2 orders, 3 products
$ python app.py
1 profile point read : Ana Lopez
2 order history 1 partition : ['o2', 'o1']
3 order + items point read : [('Kettle', 1), ('Mug', 2)]
4 product page point read : Kettle 29.5
5 browse kitchen 1 partition : ['Mug', 'Kettle']
6 sell one patch : stock 11The profile is a point read: Ana Lopez. The order history is a query inside one partition, newest first: ['o2', 'o1']. One order with its items is a point read: [('Kettle', 1), ('Mug', 2)]. The product page is a point read, Kettle 29.5, and browsing kitchen is a query inside one partition: ['Mug', 'Kettle'].
Selling a kettle is a patch with incr and a filter, c.stock > 0. The stock goes from 12 to stock 11 without reading the item first, and never below zero.
Nothing here fans out to every partition.
If a new pattern needs a different key, like orders by day for a report, I feed a second container from the change feed instead of bending this one.
Key points to remember
- Model from the access patterns, not the tables
- Embed what is read together and bounded; reference what grows or is shared
- Unbounded arrays hit the 2 MB item limit: one item per line, same partition
- Sync denormalized copies from the change feed with a saved token
- Several types in one container when they share the partition key
The full script
This is a Python script that writes every file and runs every command above in order against the Cosmos DB emulator: run pip install azure-cosmos first, and set COSMOS_ENDPOINT and COSMOS_KEY.
Show the full script that writes the files and runs every step
# ========================================================================
# Cosmos DB Interview Q&A #4 - Data modeling
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Cosmos DB Linux emulator (vNext, Docker image)
# with azure-cosmos 4.17.1; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
# 1. docker run -p 8081:8081 -p 1234:1234 mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-latest
# and wait until its log says "Now listening on: http://0.0.0.0:8081".
# 2. pip install azure-cosmos
# 3. Set COSMOS_ENDPOINT=http://localhost:8081 and COSMOS_KEY to the emulator key
# (C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==),
# then run python CosmosDB_Interview_QA_04_Code.py. The scripts create their own databases and containers.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any local data
# files) there, and a second run on old data gives different output.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('common.py', """\
import os
from azure.cosmos import CosmosClient
client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")
""")
# Step 1: An order with embedded lines vs one that references products: count the calls
write('load.py', """\
from azure.cosmos import PartitionKey
from common import db
orders = db.create_container_if_not_exists(
"orders", partition_key=PartitionKey(path="/customerId"))
products = db.create_container_if_not_exists(
"products", partition_key=PartitionKey(path="/category"))
catalog = [("p1", "kitchen", "Kettle", 29.5),
("p2", "kitchen", "Mug", 7.5),
("p3", "pantry", "Tea", 4.25)]
for pid, cat, name, price in catalog:
products.upsert_item({"id": pid, "category": cat,
"name": name, "price": price})
qty = {"p1": 1, "p2": 2, "p3": 3}
# A: embedded - each line carries what the order page shows
orders.upsert_item({"id": "o1", "customerId": "c1", "lines": [
{"productId": p, "name": n, "price": pr, "qty": qty[p]}
for p, _, n, pr in catalog]})
# B: referenced - each line keeps only the product key
orders.upsert_item({"id": "o2", "customerId": "c1", "lines": [
{"productId": p, "category": c, "qty": qty[p]}
for p, c, _, _ in catalog]})
print("products: 3, orders: o1 embedded, o2 referenced")
""")
write('reads.py', """\
from common import db
orders = db.get_container_client("orders")
products = db.get_container_client("products")
calls = 1 # A: embedded
order = orders.read_item("o1", partition_key="c1")
lines = [(l["name"], l["qty"]) for l in order["lines"]]
print("embedded :", lines, "| calls:", calls)
calls = 1 # B: referenced
order = orders.read_item("o2", partition_key="c1")
lines = []
for l in order["lines"]:
p = products.read_item(l["productId"],
partition_key=l["category"])
calls += 1
lines.append((p["name"], l["qty"]))
print("referenced:", lines, "| calls:", calls)
""")
run('python load.py')
run('python reads.py')
# Step 2: An item list that grows until the 2 MB item limit; then one item per line
write('grow.py', 'import json\nfrom azure.cosmos.exceptions import CosmosHttpResponseError\nfrom common import db\n\norders = db.get_container_client("orders")\norder = {"id": "o9", "customerId": "c9", "lines": []}\nfor rnd in [1, 2, 3]:\n start = len(order["lines"])\n order["lines"] += [{"sku": f"SKU-{start + i:06}", "qty": 1,\n "price": 9.5} for i in range(17500)]\n size = len(json.dumps(order, separators=(",", ":")))\n print(f"round {rnd}: {len(order[\'lines\'])} lines, {size} bytes")\n try:\n orders.upsert_item(order) # rewrites the whole item\n print(" saved")\n except CosmosHttpResponseError as e:\n print(" ", type(e).__name__, e.status_code)\n print(" ", str(e).split("\\n")[0])\n')
write('lines_as_items.py', """\
import json
from azure.cosmos import PartitionKey
from common import db
lines = db.create_container_if_not_exists(
"order_lines", partition_key=PartitionKey(path="/orderId"))
lines.upsert_item({"id": "o9", "orderId": "o9", "type": "header",
"customerId": "c9", "lineCount": 500})
for i in range(500): # one small item per line
lines.upsert_item({"id": f"o9-{i:04}", "orderId": "o9",
"type": "line", "sku": f"SKU-{i:06}",
"qty": 1, "price": 9.5})
q = "SELECT VALUE COUNT(1) FROM c WHERE c.type = 'line'"
print("line items in partition o9:",
list(lines.query_items(q, partition_key="o9"))[0])
pager = lines.query_items("SELECT c.sku FROM c WHERE c.type = 'line'",
partition_key="o9", max_item_count=100)
page = list(next(pager.by_page()))
print("first page:", len(page), "lines, from", page[0]["sku"])
one = lines.read_item("o9-0000", partition_key="o9")
body = {k: v for k, v in one.items() if not k.startswith("_")}
size = len(json.dumps(body, separators=(",", ":")))
print("one line item:", size, "bytes")
""")
run('python grow.py')
run('python lines_as_items.py')
# Step 3: A customer name copied into reviews, kept in sync from the change feed
write('seed.py', """\
from azure.cosmos import PartitionKey
from common import db
customers = db.create_container_if_not_exists(
"customers", partition_key=PartitionKey(path="/customerId"))
reviews = db.create_container_if_not_exists(
"reviews", partition_key=PartitionKey(path="/productId"))
for cid, name in [("c2", "Ana Lopez"), ("c3", "Raj Patel")]:
customers.upsert_item({"id": cid, "customerId": cid,
"name": name})
for rid, pid, cid, name in [("r1", "p1", "c2", "Ana Lopez"),
("r2", "p2", "c2", "Ana Lopez"),
("r3", "p3", "c2", "Ana Lopez"),
("r4", "p1", "c3", "Raj Patel")]:
reviews.upsert_item({"id": rid, "productId": pid, "stars": 5,
"customerId": cid, "customerName": name})
list(customers.query_items_change_feed(start_time="Beginning"))
token = customers.client_connection.last_response_headers["etag"]
open("customers.token", "w").write(token) # fixer starts here
print("2 customers, 4 reviews; change feed position saved")
""")
write('rename.py', """\
from common import db
customers = db.get_container_client("customers")
customers.patch_item("c2", partition_key="c2", patch_operations=[
{"op": "set", "path": "/name", "value": "Ana Lopez-Garcia"}])
print("customer c2 renamed")
""")
write('fixer.py', """\
from common import db
customers = db.get_container_client("customers")
reviews = db.get_container_client("reviews")
feed = customers.query_items_change_feed(
continuation=open("customers.token").read())
changes = list(feed) # customers changed since
token = customers.client_connection.last_response_headers["etag"]
Q = ("SELECT c.id, c.productId FROM c WHERE c.customerId = @id"
" AND c.customerName != @name")
for cust in changes:
args = [{"name": "@id", "value": cust["customerId"]},
{"name": "@name", "value": cust["name"]}]
stale = list(reviews.query_items(
Q, parameters=args, enable_cross_partition_query=True))
for r in stale: # copies to fix
reviews.patch_item(r["id"], partition_key=r["productId"],
patch_operations=[{"op": "set", "path": "/customerName",
"value": cust["name"]}])
print("change:", cust["id"], cust["name"], "| reviews patched:",
sorted(r["id"] for r in stale))
print("changes read:", len(changes))
open("customers.token", "w").write(token) # save after the work
""")
write('check.py', """\
from common import db
reviews = db.get_container_client("reviews")
rows = reviews.query_items("SELECT c.id, c.customerName FROM c",
enable_cross_partition_query=True)
for r in sorted(rows, key=lambda r: r["id"]):
print(r["id"], r["customerName"])
""")
run('python seed.py')
run('python rename.py')
run('python fixer.py')
run('python check.py')
run('python fixer.py')
# Step 4: Customers, carts and orders in one container with a type field
write('one_container.py', """\
from azure.cosmos import PartitionKey
from common import db
data = db.create_container_if_not_exists(
"customer_data", partition_key=PartitionKey(path="/customerId"))
for d in [{"id": "c5", "type": "customer", "name": "Lena Kim"},
{"id": "cart-c5", "type": "cart", "items": 2},
{"id": "o51", "type": "order", "total": 36.5},
{"id": "o52", "type": "order", "total": 12.25}]:
data.upsert_item(dict(d, customerId="c5"))
c6 = {"id": "c6", "customerId": "c6", "type": "customer",
"name": "Omar Haddad"}
o61 = {"id": "o61", "customerId": "c6", "type": "order",
"total": 8.75}
data.execute_item_batch([("upsert", (c6,)), ("upsert", (o61,))],
partition_key="c6") # all or nothing
Q = "SELECT VALUE COUNT(1) FROM c WHERE c.type = @t"
for t in ["customer", "cart", "order"]:
n = list(data.query_items(Q, enable_cross_partition_query=True,
parameters=[{"name": "@t", "value": t}]))[0]
print(f"type = {t:9} -> {n}")
rows = data.query_items("SELECT c.type, c.id FROM c ORDER BY c.type",
partition_key="c5") # one partition
print("c5:", [(r["type"], r["id"]) for r in rows])
""")
run('python one_container.py')
# Step 5: Many-to-many: products and collections with a reference list on both sides
write('link.py', """\
from azure.cosmos import PartitionKey
from common import db
products = db.get_container_client("products")
collections = db.create_container_if_not_exists(
"collections", partition_key=PartitionKey(path="/id"))
# side 1: each product lists its collections
for pid, cat, ids in [("p1", "kitchen", ["gifts", "home"]),
("p2", "kitchen", ["gifts"]),
("p3", "pantry", ["gifts"])]:
products.patch_item(pid, partition_key=cat, patch_operations=[
{"op": "set", "path": "/collectionIds", "value": ids}])
# side 2: each collection lists its products (id + partition key)
collections.upsert_item({"id": "gifts", "products": [
["p1", "kitchen"], ["p2", "kitchen"], ["p3", "pantry"]]})
collections.upsert_item({"id": "home", "products": [
["p1", "kitchen"]]})
print("linked: 3 products, 2 collections")
""")
write('browse.py', """\
from common import db
products = db.get_container_client("products")
collections = db.get_container_client("collections")
gifts = collections.read_item("gifts", partition_key="gifts")
keys = [tuple(k) for k in gifts["products"]]
found = products.read_items(keys) # one call, three items
print("collection gifts ->", sorted(p["name"] for p in found))
kettle = products.read_item("p1", partition_key="kitchen")
print("product p1 ->", kettle["collectionIds"])
Q = ("SELECT VALUE c.name FROM c"
" WHERE ARRAY_CONTAINS(c.collectionIds, @g)")
rows = products.query_items(Q, parameters=[{"name": "@g",
"value": "home"}],
enable_cross_partition_query=True)
print("query, products in home ->", sorted(rows))
""")
run('python link.py')
run('python browse.py')
# Step 6: The e-commerce model: customers with embedded orders, products by category, six access patterns
write('model.py', """\
from azure.cosmos import PartitionKey
from common import client
shop = client.create_database_if_not_exists("ecommerce")
customers = shop.create_container_if_not_exists( # customer + orders
"customers", partition_key=PartitionKey(path="/customerId"))
products = shop.create_container_if_not_exists(
"products", partition_key=PartitionKey(path="/category"))
for pid, cat, name, price, stock in [
("p1", "kitchen", "Kettle", 29.5, 12),
("p2", "kitchen", "Mug", 7.5, 40),
("p3", "pantry", "Tea", 4.25, 100)]:
products.upsert_item({"id": pid, "category": cat, "name": name,
"price": price, "stock": stock})
customers.upsert_item({"id": "c1", "customerId": "c1",
"type": "customer", "name": "Ana Lopez"})
for oid, day, items in [
("o1", "2026-09-02", [["p3", "Tea", 4.25, 2]]),
("o2", "2026-10-14", [["p1", "Kettle", 29.5, 1],
["p2", "Mug", 7.5, 2]])]:
customers.upsert_item({"id": oid, "customerId": "c1",
"type": "order", "date": day, "items": [dict(zip(
["productId", "name", "price", "qty"], i)) for i in items]})
print("ecommerce: 1 customer, 2 orders, 3 products")
""")
write('app.py', """\
from common import client
shop = client.get_database_client("ecommerce")
customers = shop.get_container_client("customers")
products = shop.get_container_client("products")
me = customers.read_item("c1", partition_key="c1")
print("1 profile point read :", me["name"])
Q = ("SELECT VALUE c.id FROM c WHERE c.type = 'order'"
" ORDER BY c.date DESC")
mine = list(customers.query_items(Q, partition_key="c1"))
print("2 order history 1 partition :", mine)
o2 = customers.read_item("o2", partition_key="c1")
print("3 order + items point read :",
[(i["name"], i["qty"]) for i in o2["items"]])
p = products.read_item("p1", partition_key="kitchen")
print("4 product page point read :", p["name"], p["price"])
Q = "SELECT VALUE c.name FROM c ORDER BY c.price"
print("5 browse kitchen 1 partition :",
list(products.query_items(Q, partition_key="kitchen")))
after = products.patch_item("p1", partition_key="kitchen",
patch_operations=[{"op": "incr", "path": "/stock", "value": -1}],
filter_predicate="FROM c WHERE c.stock > 0") # fails if sold out
print("6 sell one patch : stock", after["stock"])
""")
run('python model.py')
run('python app.py')
Part 5: Queries and Indexing
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the official Azure Cosmos DB Linux emulator (vNext) with azure-cosmos 4.17.1. Features the emulator does not support are shown as code. Part of the Real Time Azure Cosmos DB Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Azure Cosmos DB documentation (learn.microsoft.com/azure/cosmos-db) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.