Cosmos DB Interview Questions and Answers Part 8 - Transactions, Concurrency and Server-Side Logic

Cosmos DB Interview Q&A · Part 8 of 10

In this post we are going to cover transactions, concurrency and server-side logic 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: a ticketing site must never sell the same seat twice.

Cosmos DB has no locks. Instead it gives us ETags to detect that an item changed, transactions inside one logical partition, and patch to change single fields on the server. Interviewers ask about this because the bugs here are silent: a lost update or a double booking doesn't throw an error, it just leaves wrong data behind.

Part 8 of the Real Time Azure Cosmos DB Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the project
3. Question 1: Two users update the same document at the same time and one change is lost. How does optimistic concurrency with ETags help?
4. Question 2: A teammate asks whether Cosmos DB supports transactions. What is the scope?
5. Question 3: Several items in one partition must be written all or nothing. How does transactional batch work?
6. Question 4: Someone asks about stored procedures, triggers and user-defined functions. When do you use them?
7. Question 5: A counter document is updated by many writers. How do you update it safely?
8. Question 6: A teammate wants to change one field without sending the whole document. How does patch work?
9. Scenario 7: A ticketing site must never sell the same seat twice. How do you design it with Cosmos DB?
10. The full script

The short answer

Say this first

Cosmos DB gives ACID transactions within one logical partition: transactional batch (up to 100 operations, 2 MB, 5 s), stored procedures and triggers. For concurrent writers, use optimistic concurrency: send the ETag and get 412 if the item changed. Patch changes fields atomically on the server, and create-only ids or conditional patches make double bookings impossible.

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 creates the shop database. It has a container helper with a partition key path, /customerId by default, an order helper, and two small functions that print an error as its status code plus the first line, and a batch error as its status, error_index and every operation's status code.

common.py

import os
from azure.cosmos import CosmosClient, PartitionKey

client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
                      credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")


def container(name, key="/customerId"):
    return db.create_container_if_not_exists(
        name, partition_key=PartitionKey(path=key))


def order(oid, customer, total):
    return {"id": oid, "customerId": customer, "total": total}


def first_line(e):              # status + first line of the error
    return f"{e.status_code} {str(e).splitlines()[0]}"


def batch_error(e):             # CosmosBatchOperationError
    codes = [r["statusCode"] for r in e.operation_responses]
    return f"{e.status_code}, error_index {e.error_index}, {codes}"

Question 1: Two users update the same document at the same time and one change is lost. How does optimistic concurrency with ETags help?

Every item has an ETag that changes on every write. Send the ETag you read with your replace, and the write applies only if the item hasn't changed since. If it has, you get 412, precondition failed, and nothing is overwritten.

Let's see the problem first. Two agents, Anna and Ben, open customer c1. Anna saves a new phone. Ben saves a new email from his older copy, which puts the old phone back. lost.py does it without ETags, and etag.py does the same with them. We run python lost.py and python etag.py.

lost.py

from common import container

customers = container("customers")
customers.upsert_item({"id": "c1", "customerId": "c1",
                       "phone": "555-0100", "email": "ann@old.test"})
anna = customers.read_item("c1", partition_key="c1")    # agent Anna
ben = customers.read_item("c1", partition_key="c1")     # agent Ben

anna["phone"] = "555-0199"
customers.replace_item("c1", anna)
ben["email"] = "ann@new.test"
customers.replace_item("c1", ben)              # Ben's copy is older

d = customers.read_item("c1", partition_key="c1")
print("no etag:   phone", d["phone"], "| email", d["email"])

etag.py

from azure.core import MatchConditions
from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container, first_line

customers = container("customers")
customers.upsert_item({"id": "c1", "customerId": "c1",
                       "phone": "555-0100", "email": "ann@old.test"})
SAME = MatchConditions.IfNotModified
anna = customers.read_item("c1", partition_key="c1")
ben = customers.read_item("c1", partition_key="c1")

anna["phone"] = "555-0199"
customers.replace_item("c1", anna, etag=anna["_etag"],
                       match_condition=SAME)
print("Anna: saved")
ben["email"] = "ann@new.test"
try:
    customers.replace_item("c1", ben, etag=ben["_etag"],
                           match_condition=SAME)
except CosmosAccessConditionFailedError as e:
    print("Ben: ", first_line(e))
    ben = customers.read_item("c1", partition_key="c1")    # re-read
    ben["email"] = "ann@new.test"
    customers.replace_item("c1", ben, etag=ben["_etag"],
                           match_condition=SAME)
    print("Ben:  re-read and saved")
d = customers.read_item("c1", partition_key="c1")
print("with etag: phone", d["phone"], "| email", d["email"])

Terminal

$ python lost.py
no etag:   phone 555-0100 | email ann@new.test
$ python etag.py
Anna: saved
Ben:  412 (PreconditionFailed) E-tag mismatch detected
Ben:  re-read and saved
with etag: phone 555-0199 | email ann@new.test

Without ETags we end with phone 555-0100 | email ann@new.test. Ben's email is there, but Anna's phone change is lost.

With ETags, Anna's replace succeeds. Ben's carries the old ETag, so it fails with 412 (PreconditionFailed) E-tag mismatch detected. Ben re-reads, applies his email, and saves. Now we get phone 555-0199 | email ann@new.test: both changes are kept.

In Python, pass etag with match_condition=MatchConditions.IfNotModified. MatchConditions comes from azure.core, not azure.cosmos. Without it, a replace doesn't check.

Real project

Nothing is locked, conflicts are just detected. Wrap it in a small loop: read, change, replace with the ETag, and on 412, read again.

Question 2: A teammate asks whether Cosmos DB supports transactions. What is the scope?

Yes: ACID transactions with snapshot isolation, but only within one logical partition, meaning one partition key value. Transactional batch, stored procedures and triggers all run inside that scope. There's no transaction across partition keys or containers.

What	Transactional?	Scope
one item write	always atomic	the item
transactional batch	ACID, all or nothing	one partition key value
stored procedure, trigger	ACID, exception = rollback	one logical partition
across partition keys or containers	no transaction	change feed + idempotent steps

Let's test the scope. scope.py sends one batch for customer c1 that also tries to create an order for customer c2. We run python scope.py.

scope.py

from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container, order, batch_error

orders = container("orders")
ops = [("create", (order("o1", "c1", 20.5),)),       # customer c1
       ("create", (order("o2", "c2", 12.5),))]       # customer c2
try:
    orders.execute_item_batch(ops, partition_key="c1")
except CosmosBatchOperationError as e:
    print("batch for c1 and c2:", batch_error(e))

rows = orders.query_items("SELECT VALUE COUNT(1) FROM c",
                          enable_cross_partition_query=True)
print("orders written:", list(rows))

Terminal

$ python scope.py
batch for c1 and c2: 400, error_index 1, [424, 400]
orders written: [0]

The emulator rejects it: status 400, error_index 1, so the second operation is the bad request, and the first gets 424, failed dependency. And orders written: [0].

So the real design question is the partition key:

• Same partition key - order + lines, cart + items
• Across partitions - change feed, idempotent steps
• One document - is always atomic

Things that must change together should share one partition key value. Across partitions, design for eventual consistency: write the first item, let the change feed drive the next step, and make every step idempotent, so a retry is safe.

Tip

A single item write is always atomic. Many so-called transactions are really one document: embedding can remove the need for a batch.

Question 3: Several items in one partition must be written all or nothing. How does transactional batch work?

You send a list of point operations, like create, upsert, replace, patch, read and delete, all with the same partition key, as one request. They succeed or fail together.

Limit	Value
operations per batch	100
payload	2 MB
execution time	5 seconds

Let's run two batches. Batch one creates order o1 and increments the order count on customer document c1, in one partition. Batch two creates order o2, increments the count again, and then tries to create o1, which already exists. We run python batch.py.

batch.py

from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container, order, batch_error

shop = container("orders")
shop.upsert_item({"id": "c1", "customerId": "c1", "orderCount": 0})
COUNT_UP = [{"op": "incr", "path": "/orderCount", "value": 1}]

ok = shop.execute_item_batch([
    ("create", (order("o1", "c1", 20.5),)),
    ("patch", ("c1", COUNT_UP))], partition_key="c1")
print("batch 1:", [r["statusCode"] for r in ok])

try:
    shop.execute_item_batch([
        ("create", (order("o2", "c1", 9.5),)),
        ("patch", ("c1", COUNT_UP)),
        ("create", (order("o1", "c1", 1.5),))],      # o1 exists
        partition_key="c1")
except CosmosBatchOperationError as e:
    print("batch 2:", batch_error(e))

ids = shop.query_items("SELECT VALUE c.id FROM c", partition_key="c1")
print("items for c1:", sorted(ids), "| orderCount",
      shop.read_item("c1", partition_key="c1")["orderCount"])

Terminal

$ python batch.py
batch 1: [201, 200]
batch 2: 409, error_index 2, [424, 424, 409]
items for c1: ['c1', 'o1'] | orderCount 1

Batch one prints [201, 200]: created, and OK. Batch two fails with 409, error_index 2, [424, 424, 409]. The third operation fails with 409, conflict, the other two get 424, and everything rolls back. We get items for c1: ['c1', 'o1'] | orderCount 1: no o2, and the count is still 1.

In Python, a failed batch raises CosmosBatchOperationError with error_index and operation_responses, every operation's response. Log the index: it tells you which operation broke the batch.

Batch vs stored procedure

A batch needs no JavaScript, and Microsoft quotes up to 30 percent lower latency. It's the default choice for multi-item writes.

Question 4: Someone asks about stored procedures, triggers and user-defined functions. When do you use them?

All three are JavaScript running inside Cosmos DB. A stored procedure is a transaction in one logical partition: throw an exception and it all rolls back. A trigger runs before or after a write, only when the request names it. A user-defined function works only inside queries.

Server-side JavaScript	Runs when	Use it for
stored procedure	you execute it, with the partition key	a transaction with logic, in one partition
pre / post trigger	only if the write names it	stamping or checking a write
user-defined function	only inside a query	a calculation in SELECT or WHERE

Let's try a stored procedure that creates an order. sproc.py registers addOrder.js on the orders container, lists it, and then executes it. We run python sproc.py.

addOrder.js

function addOrder(order) {
    var ctx = getContext();
    var coll = ctx.getCollection();
    var ok = coll.createDocument(coll.getSelfLink(), order,
        function (err, doc) {
            if (err) throw err;              // throw = roll back
            ctx.getResponse().setBody(doc.id);
        });
    if (!ok) throw new Error("not accepted, try again");
}

sproc.py

from azure.cosmos.exceptions import CosmosHttpResponseError
from common import container, order, first_line

orders = container("orders")
orders.scripts.create_stored_procedure(
    {"id": "addOrder", "body": open("addOrder.js").read()})
print("registered:",
      [s["id"] for s in orders.scripts.list_stored_procedures()])
try:
    result = orders.scripts.execute_stored_procedure(
        "addOrder", partition_key="c1",
        params=[order("o9", "c1", 30.5)])
    print("created", result)
except CosmosHttpResponseError as e:
    print("execute:")
    print("  ", first_line(e))

Terminal

$ python sproc.py
registered: ['addOrder']
execute:
   400 (BadRequest) Server-side script execution is not supported in this emulator

The emulator accepts it and lists ['addOrder']. But execute fails: 400 (BadRequest) Server-side script execution is not supported in this emulator. On Azure, it runs.

Triggers and functions are code only here. A pre-trigger stamps a created time, but only for writes that pass pre_trigger_include:

trigger (Azure only, not run)

// pre-trigger: runs only when the write names it
function stampCreated() {
    var req = getContext().getRequest();
    var doc = req.getBody();
    doc.createdAt = new Date().toISOString();
    req.setBody(doc);
}

# Python: name the trigger on the write
orders.create_item(doc, pre_trigger_include="stampCreated")

And a tax function is used inside a SELECT:

UDF (Azure only, not run)

// UDF: called only inside a query
function tax(total) {
    return total * 0.08;
}

# Python: use it in SELECT
q = "SELECT c.id, udf.tax(c.total) AS tax FROM c"
orders.query_items(q, partition_key="c1")

Mind the limits:

• 5 second timeout - stop early, continue later
• Needs the partition key - one logical partition
• Write-heavy - not for big reads

Real project

Reach for transactional batch first. Server-side code is harder to test, and the local emulator can't run it.

Question 5: A counter document is updated by many writers. How do you update it safely?

Never read, add one, and write back without a check. Use a patch increment, which the server applies atomically, or a replace with the ETag and a retry on 412.

Let's race them. counter.py runs 20 threads that make 200 updates to each of three counters. The first reads, adds one and replaces with no ETag. The second replaces with the ETag and retries on each 412. The third sends a patch increment, with no read at all. We run python counter.py.

counter.py

from concurrent.futures import ThreadPoolExecutor
from azure.core import MatchConditions
from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container

views = container("counters", key="/id")
INCR = [{"op": "incr", "path": "/views", "value": 1}]

def no_etag(n):                        # read, add one, write back
    d = views.read_item(n, n)
    d["views"] += 1
    views.replace_item(n, d)

def etag_retry(n):                     # replace only if unchanged
    while True:
        d = views.read_item(n, n)
        d["views"] += 1
        try:
            return views.replace_item(n, d, etag=d["_etag"],
                match_condition=MatchConditions.IfNotModified)
        except CosmosAccessConditionFailedError:
            continue                   # 412: someone was faster

def patch_incr(n):                     # atomic on the server
    views.patch_item(n, n, patch_operations=INCR)

for fn in [no_etag, etag_retry, patch_incr]:
    n = fn.__name__
    views.upsert_item({"id": n, "views": 0})
    with ThreadPoolExecutor(max_workers=20) as pool:   # 20 writers
        list(pool.map(fn, [n] * 200))                   # 200 updates
    v = views.read_item(n, n)["views"]
    print(f"{n:11} {v if v == 200 else 'lost updates: under 200'}")

Terminal

$ python counter.py
no_etag     lost updates: under 200
etag_retry  200
patch_incr  200

no_etag ends with lost updates: under 200. The exact number varies, so we only print that. etag_retry reaches exactly 200, but every conflict costs a re-read and a retry. patch_incr also lands on 200.

Approach	Result	Use for
no etag	lost updates	never
etag + retry	correct, retries	logic on current state
patch incr	correct, one call	counters

So patch increment is the clean answer for counters. Keep ETag retries for changes that depend on reading the current state.

Pitfall

A hot counter is one logical partition, capped at 10,000 RU/s. For very hot counts, spread them over several documents and sum them.

Question 6: A teammate wants to change one field without sending the whole document. How does patch work?

Patch sends only the operations, up to 10 per call. The server applies them atomically and returns the full item.

Operation	What it does
set / add	set a value; creates the field if missing (add also inserts into arrays)
replace / remove	change or delete a field that must exist
incr	add a number; creates the field if missing
move	move a value from one path to another

Set creates a missing field, and replace fails on one. Let's try it. patch.py gives product p1 four operations in one call: set the price, increment the stock by minus three, add classic to the tags, and remove the promo field. Then it tries a conditional patch with a filter predicate: sell 10 if stock is at least 10, then sell 5 if stock is at least 5. We run python patch.py.

patch.py

from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container, first_line

products = container("products", key="/category")
products.upsert_item({"id": "p1", "category": "games",
                      "name": "Chess", "price": 25.5, "stock": 10,
                      "tags": ["board"], "promo": "summer"})
ops = [{"op": "set", "path": "/price", "value": 22.5},
       {"op": "incr", "path": "/stock", "value": -3},
       {"op": "add", "path": "/tags/-", "value": "classic"},
       {"op": "remove", "path": "/promo"}]
p = products.patch_item("p1", "games", patch_operations=ops)
print("price", p["price"], "| stock", p["stock"], "| tags", p["tags"],
      "| has promo:", "promo" in p)

def sell(qty):                         # only if enough stock is left
    ops = [{"op": "incr", "path": "/stock", "value": -qty}]
    return products.patch_item("p1", "games", patch_operations=ops,
        filter_predicate=f"FROM c WHERE c.stock >= {qty}")

for qty in [10, 5]:
    try:
        print(f"sell {qty}: stock now", sell(qty)["stock"])
    except CosmosAccessConditionFailedError as e:
        print(f"sell {qty}:", first_line(e))

Terminal

$ python patch.py
price 22.5 | stock 7 | tags ['board', 'classic'] | has promo: False
sell 10: 412 (PreconditionFailed) One of the specified pre-condition is not met
sell 5: stock now 2

Back come price 22.5, stock 7, two tags, ['board', 'classic'], and has promo: False. Selling 10 fails with 412 (PreconditionFailed): only seven are left. Selling 5 succeeds, leaving stock now 2.

Patch saves bandwidth, but the bigger win is concurrency. Two writers patching different fields don't overwrite each other, as full replaces would. Across write regions, non-conflicting patches merge by path.

Pitfall

Don't try to patch the id or the partition key. An item's partition key value can't change: create a new item and delete the old one.

Scenario 7: A ticketing site must never sell the same seat twice. How do you design it with Cosmos DB?

Make the seat the id: one document per sold seat, in a container partitioned by event id. Ids are unique within a logical partition, so a second create of the same seat fails with 409, conflict.

Booking is one transactional batch in the event's partition: create the seat document and create the order. If the seat create fails, the order is rolled back with it. So the e1 partition holds seat A12 plus order-A12-ann.

Let's run it. Ann books seat A12, then Ben tries A12 too. Then ten buyers race for seat B7 at the same moment. We run python seats.py.

seats.py

from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container

seats = container("seats", key="/eventId")     # a partition per event

def book(seat, buyer):                          # seat id = the seat
    seat_doc = {"id": seat, "eventId": "e1", "buyer": buyer}
    order = {"id": f"order-{seat}-{buyer}", "eventId": "e1"}
    try:
        seats.execute_item_batch([("create", (seat_doc,)),
                                  ("create", (order,))],
                                 partition_key="e1")
        return "sold"
    except CosmosBatchOperationError as e:
        codes = [r["statusCode"] for r in e.operation_responses]
        return f"rejected {codes}"

print("ann books A12:", book("A12", "ann"))
print("ben books A12:", book("A12", "ben"))
fans = [f"fan{i}" for i in range(10)]
with ThreadPoolExecutor(max_workers=10) as pool:      # same moment
    results = Counter(pool.map(lambda b: book("B7", b), fans))
print("10 buyers race for B7:", sorted(results.items()))
q = "SELECT VALUE COUNT(1) FROM c WHERE STARTSWITH(c.id, 'order-')"
print("orders in e1:", list(seats.query_items(q, partition_key="e1")))

Terminal

$ python seats.py
ann books A12: sold
ben books A12: rejected [409, 424]
10 buyers race for B7: [('rejected [409, 424]', 9), ('sold', 1)]
orders in e1: [2]

Ann gets sold. Ben gets rejected [409, 424]: 409 for the seat and 424 for his order. In the race for B7, exactly one buyer is sold and 9 are rejected, and orders in e1: [2].

The alternative is a document per seat with a status, and a conditional patch whose filter predicate says the seat is free. Losers get 412.

Design	Second buyer
create seat id (batch)	409 Conflict
conditional patch: status free	412 Precondition Failed
query, then write	both succeed: sold twice

Pitfall

Query that the seat is free, then write. Two buyers both see it free, and both buy it. The check must be the write itself.

Key points to remember

• ETag + IfNotModified: a stale write gets 412 instead of a lost update
• Transactions are ACID within one logical partition only
• Transactional batch: same partition key, 100 ops, 2 MB, 5 s; failure = 424 for the rest
• Stored procedures, triggers, UDFs: JavaScript on Azure; batch first
• Patch incr for counters; create-only ids or conditional patch for seats

The full script

This is a Python script that writes every file and runs every command above in order against the Cosmos DB emulator: start the emulator, run pip install azure-cosmos first, set COSMOS_ENDPOINT and COSMOS_KEY, and run it in an empty folder on a freshly started emulator.

Show the full script that writes the files and runs every step

Full script - writes the files and runs every step against the Cosmos DB emulator (pip install azure-cosmos), run it with Python

# ========================================================================
# Cosmos DB Interview Q&A #8 - Transactions, concurrency and server-side logic
# 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_08_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, PartitionKey

client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
                      credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")


def container(name, key="/customerId"):
    return db.create_container_if_not_exists(
        name, partition_key=PartitionKey(path=key))


def order(oid, customer, total):
    return {"id": oid, "customerId": customer, "total": total}


def first_line(e):              # status + first line of the error
    return f"{e.status_code} {str(e).splitlines()[0]}"


def batch_error(e):             # CosmosBatchOperationError
    codes = [r["statusCode"] for r in e.operation_responses]
    return f"{e.status_code}, error_index {e.error_index}, {codes}"
""")

# Step 1: Lost update without an etag, then replace with etag and IfNotModified: 412 and a retry
write('lost.py', """\
from common import container

customers = container("customers")
customers.upsert_item({"id": "c1", "customerId": "c1",
                       "phone": "555-0100", "email": "ann@old.test"})
anna = customers.read_item("c1", partition_key="c1")    # agent Anna
ben = customers.read_item("c1", partition_key="c1")     # agent Ben

anna["phone"] = "555-0199"
customers.replace_item("c1", anna)
ben["email"] = "ann@new.test"
customers.replace_item("c1", ben)              # Ben's copy is older

d = customers.read_item("c1", partition_key="c1")
print("no etag:   phone", d["phone"], "| email", d["email"])
""")
write('etag.py', """\
from azure.core import MatchConditions
from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container, first_line

customers = container("customers")
customers.upsert_item({"id": "c1", "customerId": "c1",
                       "phone": "555-0100", "email": "ann@old.test"})
SAME = MatchConditions.IfNotModified
anna = customers.read_item("c1", partition_key="c1")
ben = customers.read_item("c1", partition_key="c1")

anna["phone"] = "555-0199"
customers.replace_item("c1", anna, etag=anna["_etag"],
                       match_condition=SAME)
print("Anna: saved")
ben["email"] = "ann@new.test"
try:
    customers.replace_item("c1", ben, etag=ben["_etag"],
                           match_condition=SAME)
except CosmosAccessConditionFailedError as e:
    print("Ben: ", first_line(e))
    ben = customers.read_item("c1", partition_key="c1")    # re-read
    ben["email"] = "ann@new.test"
    customers.replace_item("c1", ben, etag=ben["_etag"],
                           match_condition=SAME)
    print("Ben:  re-read and saved")
d = customers.read_item("c1", partition_key="c1")
print("with etag: phone", d["phone"], "| email", d["email"])
""")
run('python lost.py')
run('python etag.py')

# Step 2: A batch that spans two partition key values is rejected and writes nothing
write('scope.py', """\
from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container, order, batch_error

orders = container("orders")
ops = [("create", (order("o1", "c1", 20.5),)),       # customer c1
       ("create", (order("o2", "c2", 12.5),))]       # customer c2
try:
    orders.execute_item_batch(ops, partition_key="c1")
except CosmosBatchOperationError as e:
    print("batch for c1 and c2:", batch_error(e))

rows = orders.query_items("SELECT VALUE COUNT(1) FROM c",
                          enable_cross_partition_query=True)
print("orders written:", list(rows))
""")
run('python scope.py')

# Step 3: Transactional batch: one succeeds, one fails at index 2 and rolls back
write('batch.py', """\
from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container, order, batch_error

shop = container("orders")
shop.upsert_item({"id": "c1", "customerId": "c1", "orderCount": 0})
COUNT_UP = [{"op": "incr", "path": "/orderCount", "value": 1}]

ok = shop.execute_item_batch([
    ("create", (order("o1", "c1", 20.5),)),
    ("patch", ("c1", COUNT_UP))], partition_key="c1")
print("batch 1:", [r["statusCode"] for r in ok])

try:
    shop.execute_item_batch([
        ("create", (order("o2", "c1", 9.5),)),
        ("patch", ("c1", COUNT_UP)),
        ("create", (order("o1", "c1", 1.5),))],      # o1 exists
        partition_key="c1")
except CosmosBatchOperationError as e:
    print("batch 2:", batch_error(e))

ids = shop.query_items("SELECT VALUE c.id FROM c", partition_key="c1")
print("items for c1:", sorted(ids), "| orderCount",
      shop.read_item("c1", partition_key="c1")["orderCount"])
""")
run('python batch.py')

# Step 4: A stored procedure is registered; executing it fails on the emulator
write('addOrder.js', """\
function addOrder(order) {
    var ctx = getContext();
    var coll = ctx.getCollection();
    var ok = coll.createDocument(coll.getSelfLink(), order,
        function (err, doc) {
            if (err) throw err;              // throw = roll back
            ctx.getResponse().setBody(doc.id);
        });
    if (!ok) throw new Error("not accepted, try again");
}
""")
write('sproc.py', """\
from azure.cosmos.exceptions import CosmosHttpResponseError
from common import container, order, first_line

orders = container("orders")
orders.scripts.create_stored_procedure(
    {"id": "addOrder", "body": open("addOrder.js").read()})
print("registered:",
      [s["id"] for s in orders.scripts.list_stored_procedures()])
try:
    result = orders.scripts.execute_stored_procedure(
        "addOrder", partition_key="c1",
        params=[order("o9", "c1", 30.5)])
    print("created", result)
except CosmosHttpResponseError as e:
    print("execute:")
    print("  ", first_line(e))
""")
run('python sproc.py')

# Step 5: A counter with 20 writers and 200 updates: no etag, etag with retry, patch incr
write('counter.py', """\
from concurrent.futures import ThreadPoolExecutor
from azure.core import MatchConditions
from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container

views = container("counters", key="/id")
INCR = [{"op": "incr", "path": "/views", "value": 1}]

def no_etag(n):                        # read, add one, write back
    d = views.read_item(n, n)
    d["views"] += 1
    views.replace_item(n, d)

def etag_retry(n):                     # replace only if unchanged
    while True:
        d = views.read_item(n, n)
        d["views"] += 1
        try:
            return views.replace_item(n, d, etag=d["_etag"],
                match_condition=MatchConditions.IfNotModified)
        except CosmosAccessConditionFailedError:
            continue                   # 412: someone was faster

def patch_incr(n):                     # atomic on the server
    views.patch_item(n, n, patch_operations=INCR)

for fn in [no_etag, etag_retry, patch_incr]:
    n = fn.__name__
    views.upsert_item({"id": n, "views": 0})
    with ThreadPoolExecutor(max_workers=20) as pool:   # 20 writers
        list(pool.map(fn, [n] * 200))                   # 200 updates
    v = views.read_item(n, n)["views"]
    print(f"{n:11} {v if v == 200 else 'lost updates: under 200'}")
""")
run('python counter.py')

# Step 6: Patch set, incr, add, remove, then a conditional patch with filter_predicate
write('patch.py', """\
from azure.cosmos.exceptions import CosmosAccessConditionFailedError
from common import container, first_line

products = container("products", key="/category")
products.upsert_item({"id": "p1", "category": "games",
                      "name": "Chess", "price": 25.5, "stock": 10,
                      "tags": ["board"], "promo": "summer"})
ops = [{"op": "set", "path": "/price", "value": 22.5},
       {"op": "incr", "path": "/stock", "value": -3},
       {"op": "add", "path": "/tags/-", "value": "classic"},
       {"op": "remove", "path": "/promo"}]
p = products.patch_item("p1", "games", patch_operations=ops)
print("price", p["price"], "| stock", p["stock"], "| tags", p["tags"],
      "| has promo:", "promo" in p)

def sell(qty):                         # only if enough stock is left
    ops = [{"op": "incr", "path": "/stock", "value": -qty}]
    return products.patch_item("p1", "games", patch_operations=ops,
        filter_predicate=f"FROM c WHERE c.stock >= {qty}")

for qty in [10, 5]:
    try:
        print(f"sell {qty}: stock now", sell(qty)["stock"])
    except CosmosAccessConditionFailedError as e:
        print(f"sell {qty}:", first_line(e))
""")
run('python patch.py')

# Step 7: Seat booking: seat and order in one batch; a second buyer gets 409; a race of 10 buyers
write('seats.py', """\
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from azure.cosmos.exceptions import CosmosBatchOperationError
from common import container

seats = container("seats", key="/eventId")     # a partition per event

def book(seat, buyer):                          # seat id = the seat
    seat_doc = {"id": seat, "eventId": "e1", "buyer": buyer}
    order = {"id": f"order-{seat}-{buyer}", "eventId": "e1"}
    try:
        seats.execute_item_batch([("create", (seat_doc,)),
                                  ("create", (order,))],
                                 partition_key="e1")
        return "sold"
    except CosmosBatchOperationError as e:
        codes = [r["statusCode"] for r in e.operation_responses]
        return f"rejected {codes}"

print("ann books A12:", book("A12", "ann"))
print("ben books A12:", book("A12", "ben"))
fans = [f"fan{i}" for i in range(10)]
with ThreadPoolExecutor(max_workers=10) as pool:      # same moment
    results = Counter(pool.map(lambda b: book("B7", b), fans))
print("10 buyers race for B7:", sorted(results.items()))
q = "SELECT VALUE COUNT(1) FROM c WHERE STARTSWITH(c.id, 'order-')"
print("orders in e1:", list(seats.query_items(q, partition_key="e1")))
""")
run('python seats.py')

Next in this series

Part 9: Security, Backup and Analytics

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.