Implement Data Operations in Python

In this exercise, you compare read and write request charges, apply conditional updates, expire an item with time to live (TTL), commit a transactional batch, and load items in bulk.

Before you start

To complete this exercise, you need an Azure subscription with permission to create resources and assign roles.

If your lab environment isn’t set up yet, follow Set up your lab environment to install Visual Studio Code, Git, the Azure CLI, and PowerShell 7.

You also need Python 3.12 or 3.13 installed.

Set up your Azure Cosmos DB resources

The core exercises reuse an account prepared with the core profile, not the two-item account from the first portal exercise. Before skipping setup, open Allfiles/Labs/Shared in PowerShell, sign in with az login, and set $resourceGroup, $location, and $accountName to your recorded values. Run ./verify.ps1 -ResourceGroup $resourceGroup -AccountName $accountName -LabProfile core and continue only when it succeeds. In Data Explorer, confirm 295 items in cosmicworks/product and 237 in cosmicworks/productMeta with SELECT VALUE COUNT(1) FROM c.

If you have no verified core account, follow the setup steps below. To add missing resources to an existing lab account, pass its explicit -AccountName to setup rather than using a different module’s name prefix. Reseeding restores canonical items but doesn’t remove extra items or reset container policies. Resolve mismatches before continuing; don’t reset a shared account automatically.

Start Visual Studio Code.
If you don’t have the lab code yet, clone the repository for DP-420: open the command palette with Ctrl+Shift+P, run Git: Clone, and enter the following URL. Choose a local folder when prompted. Otherwise, open the folder from your previous clone.
```
 https://github.com/microsoftlearning/dp-420-cosmos-db-dev
```
Once the repository is cloned, open that local folder in Visual Studio Code.
In the Explorer pane, browse to the Allfiles/Labs/Shared folder.
Open the context menu for the folder and select Open in Integrated Terminal. If the terminal isn’t PowerShell, select the dropdown beside the + in the terminal toolbar and choose PowerShell.
Sign in to the Azure CLI. A browser window opens so you can sign in to Azure.
```
 az login
```
Set variables for the resource group (If a resource group was provided by your lab environment, use that name) and region you want to use. Change either value if you prefer a different resource group name or region.
```
 $resourceGroup = "ResourceGroup1"
 $location = "westus2"
```
Run the setup script.
```
 ./setup.ps1 -ResourceGroup $resourceGroup -Location $location -NamePrefix dp420lab04 -LabProfile core
```
Azure Cosmos DB account names have to be globally unique, so the script builds one for you by adding six random characters to the prefix, giving a name like dp420lab04a7f3k9.
Wait for the script to finish. The whole script takes 5-10 minutes to run.
Record the Account name and Account endpoint values the script prints. You need the endpoint throughout this exercise, and it looks like https://<your-account-name>.documents.azure.com:443/.

The script creates the following resources, with throughput measured in request units per second (RU/s):

Resource	Configuration
Azure Cosmos DB account	API for NoSQL, with key-based authentication disabled
`cosmicworks` database	Holds every container this learning path uses
`product` container	Partitioned on `/categoryId`, autoscale up to 1,000 RU/s, seeded with 295 items
`productMeta` container	Partitioned on `/type`, autoscale up to 1,000 RU/s, seeded with 237 items
`leases` container	Partitioned on `/id`, manual throughput of 400 RU/s, empty
`operations` container	Partitioned on `/categoryId`, autoscale up to 1,000 RU/s, empty. Tasks 2 through 6 work here
`bulkload` container	Partitioned on `/categoryId`, autoscale up to 1,000 RU/s, empty. Task 7 writes to it
Role assignment	Cosmos DB Built-in Data Contributor, granted to your signed-in identity

Because key-based authentication is disabled, no key or connection string appears anywhere in this exercise. Every operation authenticates with the identity from your az login session, which is the recommended approach for new accounts.

📝 A new role assignment takes a few minutes to propagate. If a later step fails with a 403 error, wait a moment and try again.

Task 1: Set up the project and connect to the container

Enable item-level TTL on the operations container. Replace <your-account-name> with your account name:

az cosmosdb sql container update `
    --account-name <your-account-name> `
    --resource-group $resourceGroup `
    --database-name cosmicworks `
    --name operations `
    --ttl -1

A value of -1 expires only the items that set their own ttl.

Open a terminal, create a project folder, and set up a virtual environment:

 mkdir cosmos-operations-exercise
 cd cosmos-operations-exercise
 python -m venv .venv

Activate the environment. On Windows:

 .venv\Scripts\activate

On macOS or Linux:

 source .venv/bin/activate

Install the SDK packages:

 pip install azure-cosmos azure-identity

In the cosmos-operations-exercise folder, create app.py with the following code. Set endpoint to the account endpoint the setup script printed:

 import time

 from azure.core import MatchConditions
 from azure.cosmos import CosmosClient, exceptions
 from azure.identity import DefaultAzureCredential

 endpoint = "<cosmos-endpoint>"

 # Reuse one client for all operations.
 client = CosmosClient(url=endpoint, credential=DefaultAzureCredential())
 container = client.get_database_client("cosmicworks").get_container_client("operations")

 # ---- Each task replaces everything below this line ----

 print("Container ready.")

❗ For each task, replace only the code below the marker comment. Keep the imports, client setup, and endpoint above it unchanged.

Run the script and confirm the output reads Container ready.:
```
 python app.py
```

Task 2: Compare a point read against a query

Create one item, then read it back two ways and compare the request charge.

In app.py, replace everything below the marker comment with the following code. It creates the saddle item that the rest of this exercise works with:

 item_id = "027D0B9A-F9D9-4C96-8213-C8546C4AAE71"
 category_id = "26C74104-40BC-4541-8EF5-9892F7F03D72"

 saddle = {
     "id": item_id,
     "categoryId": category_id,
     "categoryName": "Components, Saddles",
     "sku": "SE-R581",
     "name": "LL Road Seat/Saddle",
     "price": 27.12,
     "tags": [
         {"id": "0573D684-9140-4DEE-89AF-4E4A90E65666", "name": "Tag-113"},
         {"id": "6C2F05C8-1E61-4912-BE1A-C67A378429BB", "name": "Tag-5"},
     ],
 }

 # Create rejects an existing ID in this partition instead of overwriting it.
 try:
     container.create_item(body=saddle)
     print("Create:      item created")
 except exceptions.CosmosResourceExistsError:
     print("Create:      item already exists")

Add the point read and print its charge:

 # Knowing both the item ID and partition key avoids executing a SQL query.
 read_item = container.read_item(item=item_id, partition_key=category_id)
 print(f"Point read:  {read_item.get_response_headers()['x-ms-request-charge']} RU")

Add an equivalent query that filters on the same two values. Use a response hook to total the query’s request charge:

 query_charge = 0.0

 def capture_charge(headers, results):
     global query_charge
     # Sum each response so the comparison covers the whole query.
     query_charge += float(headers["x-ms-request-charge"])

 # Bind values separately from SQL and explicitly route to one partition.
 # list() consumes the lazy results before the final charge is printed.
 results = list(container.query_items(
     query="SELECT * FROM c WHERE c.id = @id AND c.categoryId = @categoryId",
     parameters=[
         {"name": "@id", "value": item_id},
         {"name": "@categoryId", "value": category_id},
     ],
     partition_key=category_id,
     response_hook=capture_charge,
 ))

 print(f"Query:       {query_charge:.2f} RU")

Run the script and compare the two read charges:

 python app.py

 Create:      item created
 Point read:  1.00 RU
 Query:       2.92 RU

The point read costs less than the query because it doesn’t use the query engine. Use a point read when you know both the item ID and partition key.

Task 3: Run the write operations

Replace, upsert, and delete the item, then recreate it for the tasks that follow.

In app.py, replace everything below the marker comment with the following code. It reads the item created in Task 2 and replaces it with a new price:

 item_id = "027D0B9A-F9D9-4C96-8213-C8546C4AAE71"
 category_id = "26C74104-40BC-4541-8EF5-9892F7F03D72"

 # Read the full document so replacing it preserves fields not being changed.
 saddle = container.read_item(item=item_id, partition_key=category_id)

 saddle["price"] = 32.55

 replaced = container.replace_item(item=item_id, body=saddle)
 print(f"Replace:     {replaced.get_response_headers()['x-ms-request-charge']} RU")

A replace fails with HTTP 404 if the item doesn’t exist.

Add an upsert that suppresses the response payload:

 saddle["price"] = 30.00

 # Upsert creates or replaces. Suppressing its response body saves bandwidth,
 # not the RU cost of the write; response headers remain available.
 upserted = container.upsert_item(body=saddle, no_response=True)
 charge = upserted.get_response_headers()["x-ms-request-charge"]
 print(f"Upsert:      {charge} RU (no payload returned)")

Even with the response body suppressed, the response headers still provide the request charge.

Add a delete, then recreate the item so later tasks have something to work with:

 container.delete_item(item=item_id, partition_key=category_id)
 # Read this charge before the following create replaces the last-response headers.
 charge = container.client_connection.last_response_headers["x-ms-request-charge"]
 print(f"Delete:      {charge} RU")

 saddle["price"] = 27.12
 container.create_item(body=saddle)

Run the script and compare the write charges against the read charges from Task 2:

 Replace:     10.67 RU
 Upsert:      10.67 RU (no payload returned)
 Delete:      9.52 RU

Writes include document and index-maintenance work, so they generally cost more than point reads. Charges depend on item size and indexing policy. Suppressing the response body saves bandwidth, not RUs.

Task 4: Apply a conditional patch and guard a replace with an ETag

Patch selected properties with a price condition, then use an ETag to prevent replacing an item version that has changed.

In app.py, replace everything below the marker comment with the following code. It patches two properties, guarded by a filter predicate that tests the current price:

 item_id = "027D0B9A-F9D9-4C96-8213-C8546C4AAE71"
 category_id = "26C74104-40BC-4541-8EF5-9892F7F03D72"

 try:
     # Both patch operations apply to one item only when the predicate holds.
     patched = container.patch_item(
         item=item_id,
         partition_key=category_id,
         patch_operations=[
             {"op": "set", "path": "/name", "value": "LL Road Seat/Saddle, Clearance"},
             {"op": "incr", "path": "/price", "value": 5.00},
         ],
         filter_predicate="FROM c WHERE c.price < 100",
     )
     print(f"Patch:       {patched.get_response_headers()['x-ms-request-charge']} RU")
 except exceptions.CosmosAccessConditionFailedError:
     print("Patch:       412 Precondition Failed")

Run the script. The predicate holds, so the patch succeeds and the price is now 32.12. Record its request charge alongside the replace charge from Task 3. Patch doesn’t guarantee a lower write charge; its benefits include a smaller request payload and avoiding a preceding read when the changes are already known.

Now force the condition to fail. Append the following code, which uses a predicate the item no longer satisfies:

 # A false predicate rejects the patch without incrementing the price.
 try:
     container.patch_item(
         item=item_id,
         partition_key=category_id,
         patch_operations=[{"op": "incr", "path": "/price", "value": 5.00}],
         filter_predicate="FROM c WHERE c.price < 30",
     )
     print("Failed patch: unexpectedly succeeded")
 except exceptions.CosmosAccessConditionFailedError:
     print("Failed patch: 412 Precondition Failed, as expected")

Now guard a replace with an ETag instead. Append the following code, which reads the item, keeps its ETag, and writes with that ETag attached:

 current = container.read_item(item=item_id, partition_key=category_id)
 # IfNotModified accepts the replacement only if this item version is still current.
 etag = current["_etag"]

 current["price"] = 45.00
 container.replace_item(
     item=item_id,
     body=current,
     etag=etag,
     match_condition=MatchConditions.IfNotModified,
 )

 print("Replace:     succeeded with a current ETag")

Reuse the same, now stale, ETag on a second replace:

 # The successful replace changed the ETag; reusing the saved one must fail.
 try:
     current["price"] = 50.00
     container.replace_item(
         item=item_id,
         body=current,
         etag=etag,
         match_condition=MatchConditions.IfNotModified,
     )
     print("Stale replace: unexpectedly succeeded")
 except exceptions.CosmosAccessConditionFailedError:
     print("Stale replace: 412 Precondition Failed, as expected")

Run the script. The first replace succeeds and changes the item’s ETag, so the second one fails.

Both examples reject the write with HTTP 412 when its precondition isn’t met. Read the current item before deciding whether to reapply the change.

Task 5: Expire an item with time to live

Expire a session item after 30 seconds.

In app.py, replace everything below the marker comment with the following code. It creates a session item with a 30-second lifetime:

 category_id = "26C74104-40BC-4541-8EF5-9892F7F03D72"

 # Item TTL is in seconds from the last write. Container TTL must be enabled.
 # Upserting this item again restarts its expiration window.
 container.upsert_item(body={
     "id": "session-9f21",
     "categoryId": category_id,
     "state": "active",
     "ttl": 30,
 })
 print("Session created with a 30 second TTL.")

Add reads before and after expiration:

 def read_session(label):
     try:
         container.read_item(item="session-9f21", partition_key=category_id)
         print(f"{label} session still present")
     except exceptions.CosmosResourceNotFoundError:
         print(f"{label} session expired")

 # Read before and after the item's 30-second TTL.
 time.sleep(10)
 read_session("After 10s:")

 time.sleep(25)
 read_session("After 35s:")

Items are excluded from reads and queries once their TTL expires. Physical deletion happens later in the background.

Run the script. The first read succeeds and the second raises a not-found error:

 Session created with a 30 second TTL.
 After 10s: session still present
 After 35s: session expired

With a container TTL of -1, only items with their own positive ttl expire. The session expires; the product remains.

Task 6: Commit a transactional batch

Write two items atomically within one partition, then test a batch containing different partition keys.

In app.py, replace everything below the marker comment with the following code. Both items share the partition key value used throughout this exercise:

 category_id = "26C74104-40BC-4541-8EF5-9892F7F03D72"

 # Each tuple contains an operation name and its positional arguments.
 # All item bodies must match the batch's one partition key value.
 batch_operations = [
     ("upsert", ({"id": "201D0D79-81AD-43D2-AD6E-F09EEE6AC2D7", "categoryId": category_id,
                  "categoryName": "Components, Saddles", "sku": "SE-M798",
                  "name": "ML Mountain Seat/Saddle", "price": 39.14, "tags": []},)),
     ("upsert", ({"id": "3FE1A99E-DE14-4D11-B635-F5D39258A0B9", "categoryId": category_id,
                  "categoryName": "Components, Saddles", "sku": "SE-T924",
                  "name": "HL Touring Seat/Saddle", "price": 52.64, "tags": []},)),
 ]

 # These writes commit together or roll back together, unlike independent bulk writes.
 batch_response = container.execute_item_batch(
     batch_operations=batch_operations,
     partition_key=category_id,
 )

 for index, result in enumerate(batch_response):
     print(f"  Operation {index}: {result.get('statusCode')}")

Run the script. Every operation reports a success status, and both items exist in the container.

Now build a batch that mixes partition key values. Append the following code:

 # The second body deliberately uses a different /categoryId, invalidating
 # the entire transaction, including the otherwise valid first create.
 bad_operations = [
     ("create", ({"id": "5996B5E0-6EC7-4CB7-A924-7B5A053AE980", "categoryId": category_id,
                  "categoryName": "Components, Saddles", "sku": "SE-M236",
                  "name": "LL Mountain Seat/Saddle", "price": 27.12, "tags": []},)),
     ("create", ({"id": "47ED1C3B-C205-4507-94EE-3B69A744B261",
                  "categoryId": "14A1AD5D-59EA-4B63-A189-67B077783B0E",
                  "categoryName": "Accessories, Helmets", "sku": "HL-U509",
                  "name": "Sport-100 Helmet, Black", "price": 34.99, "tags": []},)),
 ]

 try:
     container.execute_item_batch(bad_operations, partition_key=category_id)
     print("Bad batch:   unexpectedly succeeded")
 except exceptions.CosmosBatchOperationError as e:
     print(f"Bad batch:   operation {e.error_index} failed with status {e.status_code}")

Run the script. The batch fails, and error_index identifies which operation caused it. Confirm in the Azure portal’s Data Explorer that neither item was written.

Task 7: Load items in bulk

Load 2,000 independent items and record the elapsed time.

This task loads items into bulkload.

Install the async transport dependency:
```
 pip install aiohttp
```

Create a second file named bulk_load.py with the following contents. Set the endpoint to the same value you used earlier:

 import asyncio
 import time
 import uuid
 from azure.cosmos import exceptions
 from azure.cosmos.aio import CosmosClient
 from azure.identity.aio import DefaultAzureCredential

 endpoint = "<cosmos-endpoint>"

 async def main():
     items = [
         {
             "id": str(uuid.uuid4()),
             "categoryId": f"bulk-category-{i % 20}",
             "categoryName": f"Bulk Category {i % 20}",
             "sku": f"BL-{i:05d}",
             "name": f"Bulk Product {i}",
             "price": 10.00 + i % 500,
             "tags": [],
         }
         for i in range(2000)
     ]

     # Async context managers close credential and connection resources.
     async with DefaultAzureCredential() as credential:
         async with CosmosClient(endpoint, credential) as client:
             container = client.get_database_client("cosmicworks").get_container_client("bulkload")

             # Bound concurrent independent writes; this isn't an atomic batch.
             semaphore = asyncio.Semaphore(100)

             async def create(item):
                 async with semaphore:
                     return await container.create_item(body=item)

             start = time.perf_counter()

             pending = items
             first_pass_failures = 0

             for attempt in range(3):
                 # Preserve each outcome in input order so failed items can be identified.
                 results = await asyncio.gather(
                     *(create(item) for item in pending), return_exceptions=True)

                 # Retry only explicit 429 or 449 rejections; propagate other errors.
                 for result in results:
                     if isinstance(result, BaseException) and (
                         not isinstance(result, exceptions.CosmosHttpResponseError)
                         or result.status_code not in (429, 449)
                     ):
                         raise result

                 # Only rejected items enter the next pass; successes aren't resent.
                 pending = [item for item, result in zip(pending, results)
                            if isinstance(result, Exception)]

                 if attempt == 0:
                     first_pass_failures = len(pending)

                 if not pending:
                     break

                 await asyncio.sleep(2)

             elapsed = time.perf_counter() - start

             print(f"Bulk load:   {len(items) - len(pending)} of {len(items)} written in {elapsed:.1f} seconds")
             print(f"Retried:     {first_pass_failures} on the first pass, {len(pending)} still outstanding")

 asyncio.run(main())

The semaphore limits concurrent writes to 100. The loader retries HTTP 429 and 449 responses and stops on other errors. A timed-out create might have committed, so check its outcome before retrying.

The generated items use 20 logical partition key values. A well-distributed key allows write activity to spread across physical partitions as the container scales.

Run the script and note the elapsed time:
```
 python bulk_load.py
```
```
 Bulk load:   2000 of 2000 written in 21.4 seconds
 Retried:     22 on the first pass, 0 still outstanding
```
Elapsed time and retry counts vary. Confirm that 2,000 items were written and none are outstanding. Rerunning the program generates new IDs and can add another 2,000 items.
When you finish testing, deactivate the Python environment:
```
 deactivate
```

Bulk loading improves throughput by running independent writes concurrently; it doesn’t reduce each item’s RU charge. Completion time depends on available RU/s and client-side concurrency.

Clean up resources

When you finish the course, delete the resource group only if you created it and every resource in it can be removed. If your lab provided ResourceGroup1, skip this command and delete only the exercise resources you no longer need:

az group delete --name $resourceGroup --yes --no-wait

If your lab environment provided the resource group, delete only the Azure Cosmos DB account instead:

az cosmosdb delete --name <your-account-name> --resource-group $resourceGroup --yes