Updated code example

Author: egges

2026/cqrs/EXAMPLES.md

@@ -12,4 +12,38 @@ source .venv/bin/activate
 uvicorn before:app --reload # to run the before version of the code
 ```
 
-Here are a few example requests you can use to test the FastAPI app:
+Here are a few example requests you can use to test the FastAPI app:
+
+# Create a ticket
+
+curl -X POST http://localhost:8000/tickets \
+  -H "Content-Type: application/json" \
+  -d '{
+    "customer_id": "cust-123",
+    "subject": "Login not working",
+    "message": "I cannot log in after resetting my password."
+  }'
+
+# Update ticket status
+
+curl -X PATCH http://localhost:8000/tickets/69722457531892ec7a576522 \
+  -H "Content-Type: application/json" \
+  -d '{
+    "status": "triaged"
+  }'
+
+# Update ticket status (after version)
+
+curl -X POST http://localhost:8000/tickets/697239ff4e89771fad8ed94f/status \
+  -H "Content-Type: application/json" \
+  -d '{
+    "new_status": "triaged"
+  }'
+
+# Add a note
+
+curl -X POST http://localhost:8000/tickets/697239ff4e89771fad8ed94f/agent-note \
+  -H "Content-Type: application/json" \
+  -d '{
+    "note": "Customer reset their password twice; investigating auth service."
+  }'

2026/cqrs/after.py

@@ -3,11 +3,9 @@
 from datetime import datetime, timezone
 from typing import Any, Literal
 
-from bson import ObjectId
+from db import Database, get_db, oid_str, parse_object_id, shutdown_db
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
-from pymongo import AsyncMongoClient
-from pymongo.asynchronous.database import AsyncDatabase
 
 # ----------------------------
 # Types / helpers
@@ -22,17 +20,6 @@ def utcnow() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def oid_str(oid: ObjectId) -> str:
-    return str(oid)
-
-
-def parse_object_id(ticket_id: str) -> ObjectId:
-    try:
-        return ObjectId(ticket_id)
-    except Exception as e:
-        raise HTTPException(status_code=400, detail="Invalid ticket id") from e
-
-
 def make_preview(message: str) -> str:
     msg = message.strip().replace("\n", " ")
     return msg if len(msg) <= PREVIEW_LEN else msg[: PREVIEW_LEN - 1] + "…"
@@ -42,8 +29,6 @@ def make_preview(message: str) -> str:
 # MongoDB configuration (local + auth)
 # ----------------------------
 
-MONGODB_URI = "mongodb://root:example@localhost:27017/?authSource=admin"
-DATABASE_NAME = "cqrs_demo"
 
 COMMANDS_COLL = "ticket_commands"  # source of truth
 READS_COLL = "ticket_reads"  # read projection for list/dashboard
@@ -54,16 +39,10 @@ def make_preview(message: str) -> str:
 
 app = FastAPI()
 
-mongo_client: AsyncMongoClient | None = None
-db: AsyncDatabase | None = None
-
 
 @app.on_event("startup")
 async def startup() -> None:
-    global mongo_client, db
-
-    mongo_client = AsyncMongoClient(MONGODB_URI)
-    db = mongo_client[DATABASE_NAME]
+    db = get_db()
 
     # Write-side indexes (source of truth)
     await db[COMMANDS_COLL].create_index("status")
@@ -77,13 +56,7 @@ async def startup() -> None:
 
 @app.on_event("shutdown")
 async def shutdown() -> None:
-    if mongo_client is not None:
-        mongo_client.close()
-
-
-def get_db() -> AsyncDatabase:
-    assert db is not None
-    return db
+    await shutdown_db()
 
 
 # ----------------------------
@@ -135,7 +108,7 @@ class TicketDetails(BaseModel):
 # ----------------------------
 
 
-async def cmd_create_ticket(db: AsyncDatabase, cmd: CreateTicket) -> str:
+async def cmd_create_ticket(db: Database, cmd: CreateTicket) -> str:
     now = utcnow()
     doc: dict[str, Any] = {
         "customer_id": cmd.customer_id,
@@ -150,9 +123,7 @@ async def cmd_create_ticket(db: AsyncDatabase, cmd: CreateTicket) -> str:
     return oid_str(res.inserted_id)
 
 
-async def cmd_update_status(
-    db: AsyncDatabase, ticket_id: str, cmd: UpdateStatus
-) -> None:
+async def cmd_update_status(db: Database, ticket_id: str, cmd: UpdateStatus) -> None:
     _id = parse_object_id(ticket_id)
 
     existing = await db[COMMANDS_COLL].find_one({"_id": _id})
@@ -168,9 +139,7 @@ async def cmd_update_status(
     )
 
 
-async def cmd_add_agent_note(
-    db: AsyncDatabase, ticket_id: str, cmd: AddAgentNote
-) -> None:
+async def cmd_add_agent_note(db: Database, ticket_id: str, cmd: AddAgentNote) -> None:
     _id = parse_object_id(ticket_id)
 
     existing = await db[COMMANDS_COLL].find_one({"_id": _id})
@@ -188,7 +157,7 @@ async def cmd_add_agent_note(
 # ----------------------------
 
 
-async def project_ticket(db: AsyncDatabase, ticket_id: str) -> None:
+async def project_ticket(db: Database, ticket_id: str) -> None:
     """
     Read model goal:
     - store exactly what the list/dashboard needs
@@ -345,6 +314,8 @@ async def dashboard() -> dict[str, int]:
     pipeline = [{"$group": {"_id": "$status", "count": {"$sum": 1}}}]
 
     counts: dict[str, int] = {"open": 0, "triaged": 0, "closed": 0}
-    async for row in db[READS_COLL].aggregate(pipeline):
+
+    cursor = await db[READS_COLL].aggregate(pipeline)
+    async for row in cursor:
         counts[str(row["_id"])] = int(row["count"])
     return counts

2026/cqrs/before.py

@@ -3,31 +3,20 @@
 from datetime import datetime, timezone
 from typing import Any, Literal
 
-from bson import ObjectId
-from db import get_db, shutdown_db
+from db import get_db, oid_str, shutdown_db
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
 
 Status = Literal["open", "triaged", "closed"]
 
 PREVIEW_LEN = 80
+TICKETS_COLL = "tickets"
 
 
 def utcnow() -> datetime:
     return datetime.now(timezone.utc)
 
 
-def oid_str(oid: ObjectId) -> str:
-    return str(oid)
-
-
-def parse_object_id(ticket_id: str) -> ObjectId:
-    try:
-        return ObjectId(ticket_id)
-    except Exception as e:
-        raise HTTPException(status_code=400, detail="Invalid ticket id") from e
-
-
 def make_preview(message: str) -> str:
     msg = message.strip().replace("\n", " ")
     return msg if len(msg) <= PREVIEW_LEN else msg[: PREVIEW_LEN - 1] + "…"

2026/cqrs/db.py

@@ -1,18 +1,28 @@
-from __future__ import annotations
+from typing import Any, Mapping
 
+from bson import ObjectId
 from pymongo.asynchronous.database import AsyncDatabase
+from pymongo.asynchronous.mongo_client import AsyncMongoClient
 
 MONGODB_URI = "mongodb://root:example@localhost:27017/?authSource=admin"
 DATABASE_NAME = "cqrs_demo"
-TICKETS_COLL = "tickets"
 
-mongo_client = AsyncMongoClient(MONGODB_URI)
-db: AsyncDatabase | None = None
+type Database = AsyncDatabase[Mapping[str, Any]]
+mongo_client = AsyncMongoClient[Mapping[str, Any]](MONGODB_URI)
+db: Database | None = None
 
 
-def get_db() -> AsyncDatabase:
+def get_db() -> Database:
     return mongo_client[DATABASE_NAME]
 
 
-def shutdown_db() -> None:
-    mongo_client.close()
+async def shutdown_db() -> None:
+    await mongo_client.close()
+
+
+def oid_str(oid: ObjectId) -> str:
+    return str(oid)
+
+
+def parse_object_id(ticket_id: str) -> ObjectId:
+    return ObjectId(ticket_id)