NoSQL in Microservices

from pymongo import MongoClient
import datetime, uuid

client = MongoClient("mongodb://localhost:27017/")
db     = client["product_service"]

def decrement_inventory(product_id: str, quantity: int):
    """
    Outbox pattern: write the business state change AND the outbox event
    in a single MongoDB transaction. The relay process reads the outbox
    and publishes to Kafka — separately and reliably.
    """
    with client.start_session() as session:
        session.start_transaction()
        try:
            # Step 1: update inventory
            result = db.inventory.update_one(
                {"_id": product_id, "stock": {"$gte": quantity}},
                {"$inc": {"stock": -quantity}},
                session=session
            )
            if result.modified_count == 0:
                raise ValueError("Insufficient stock")

            # Step 2: write event to outbox collection (same transaction)
            db.outbox.insert_one({
                "_id":        str(uuid.uuid4()),
                "topic":      "inventory.updated",
                "payload":    {"product_id": product_id,
                               "quantity_decremented": quantity},
                "created_at": datetime.datetime.utcnow(),
                "published":  False     # relay will set this True after Kafka ACK
            }, session=session)

            session.commit_transaction()
            return {"ok": True}

        except Exception as e:
            session.abort_transaction()
            return {"ok": False, "reason": str(e)}

import time
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def outbox_relay():
    """Poll outbox for unpublished events and forward to Kafka."""
    while True:
        # Find up to 100 unpublished events, oldest first
        pending = list(db.outbox.find(
            {"published": False},
            sort=[("created_at", 1)],
            limit=100
        ))

        for event in pending:
            # Publish to Kafka topic
            producer.send(event["topic"], value=event["payload"])
            producer.flush()   # wait for broker ACK

            # Mark as published only after Kafka confirms receipt
            db.outbox.update_one(
                {"_id": event["_id"]},
                {"$set": {"published": True,
                          "published_at": datetime.datetime.utcnow()}}
            )

        time.sleep(0.1)   # poll every 100ms

# Event store in MongoDB — append-only, never update or delete
def append_event(order_id: str, event_type: str, data: dict):
    db.order_events.insert_one({
        "order_id":   order_id,
        "event_type": event_type,
        "data":       data,
        "occurred_at": datetime.datetime.utcnow(),
        "version":    db.order_events.count_documents(
                          {"order_id": order_id}) + 1
    })

# Record the order lifecycle
append_event("ord_9912", "OrderCreated",
    {"customer_id": "cust_441", "items": ["prod_001"], "total": 149.99})
append_event("ord_9912", "PaymentConfirmed",
    {"payment_id": "pay_7721", "amount": 149.99})
append_event("ord_9912", "OrderShipped",
    {"carrier": "DHL", "tracking": "DHL998821"})

# Rebuild current state by replaying all events
def get_order_state(order_id: str) -> dict:
    events = list(db.order_events.find(
        {"order_id": order_id},
        sort=[("version", 1)]
    ))
    state = {}
    for event in events:
        if   event["event_type"] == "OrderCreated":    state.update(event["data"]); state["status"] = "created"
        elif event["event_type"] == "PaymentConfirmed": state["status"] = "paid"
        elif event["event_type"] == "OrderShipped":    state["status"] = "shipped"; state.update(event["data"])
    return state

from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
from contextlib import asynccontextmanager
import os

# ---- Startup / shutdown connection lifecycle ----
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Open pool once at startup — not per-request
    app.state.mongo = MongoClient(
        os.environ["MONGO_URI"],
        maxPoolSize=20,          # sized to pod concurrency
        minPoolSize=5,
        serverSelectionTimeoutMS=3000,
        waitQueueTimeoutMS=2000  # fail fast, don't queue forever
    )
    app.state.db = app.state.mongo["product_service"]
    yield
    # Clean shutdown — return connections to pool
    app.state.mongo.close()

app = FastAPI(lifespan=lifespan)

@app.get("/products/{product_id}")
async def get_product(product_id: str):
    doc = app.state.db.products.find_one(
        {"_id": product_id},
        {"_id": 1, "name": 1, "price": 1, "stock": 1}  # projection
    )
    if not doc:
        raise HTTPException(status_code=404, detail="Not found")
    return doc

@app.post("/products/{product_id}/purchase")
async def purchase(product_id: str, quantity: int):
    result = decrement_inventory(product_id, quantity)
    if not result["ok"]:
        raise HTTPException(status_code=409, detail=result["reason"])
    return {"purchased": True}