Caching, Queues & Event-Driven Architecture¶

import json
import redis
from typing import Optional

r = redis.Redis(host="localhost", decode_responses=True)

def get_product(product_id: int) -> Optional[dict]:
    """Cache-aside: check cache → miss → read DB → populate → return."""
    cache_key = f"product:{product_id}"

    # 1. Check cache
    cached = r.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 2. Cache miss — read from DB
    product = db_get_product(product_id)   # your DB call
    if product is None:
        return None

    # 3. Populate cache with TTL (stagger to avoid thundering herd)
    import random
    ttl = 300 + random.randint(0, 60)      # 5–6 min, jittered
    r.setex(cache_key, ttl, json.dumps(product))
    return product

def update_product(product_id: int, data: dict) -> dict:
    """On write: update DB, then invalidate the cache key."""
    product = db_update_product(product_id, data)
    r.delete(f"product:{product_id}")      # invalidate
    return product

import uuid
import time
import redis
from contextlib import contextmanager

r = redis.Redis(host="localhost", decode_responses=True)

# Atomic release: only delete if we own the lock (prevent releasing another owner's lock)
RELEASE_SCRIPT = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
"""
_release = r.register_script(RELEASE_SCRIPT)

@contextmanager
def redis_lock(key: str, timeout_seconds: int = 10, retry_delay: float = 0.1, max_retries: int = 50):
    """
    Acquire a distributed lock using SET NX EX (atomic acquire + expiry).
    Releases automatically on context exit.
    """
    lock_id = str(uuid.uuid4())           # unique token — proves we own the lock
    full_key = f"lock:{key}"
    acquired = False

    for attempt in range(max_retries):
        # SET key lock_id NX EX timeout  — atomic: set only if not exists, with expiry
        result = r.set(full_key, lock_id, nx=True, ex=timeout_seconds)
        if result:
            acquired = True
            break
        time.sleep(retry_delay)

    if not acquired:
        raise TimeoutError(f"Could not acquire lock '{key}' after {max_retries} retries")

    try:
        yield lock_id
    finally:
        _release(keys=[full_key], args=[lock_id])  # safe release

# Usage — only one worker runs the job at a time across all instances
def run_nightly_report():
    try:
        with redis_lock("nightly-report", timeout_seconds=300):
            print("Lock acquired. Running report...")
            generate_report()   # only one instance does this
    except TimeoutError:
        print("Another instance is running the report. Skipping.")

from celery import Celery
from celery.utils.log import get_task_logger

app = Celery("myapp", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")
logger = get_task_logger(__name__)

# Configure DLQ: failed tasks routed to 'dead_letter' queue after max_retries
app.conf.update(
    task_routes={
        "tasks.process_payment": {"queue": "payments"},
        "tasks.handle_dead_letter": {"queue": "dead_letter"},
    },
    task_annotations={
        "tasks.process_payment": {
            "max_retries": 5,
            "default_retry_delay": 60,
        }
    },
)

@app.task(
    bind=True,
    max_retries=5,
    acks_late=True,          # only ack after task completes (prevents loss on crash)
    reject_on_worker_lost=True,  # re-queue if the worker dies
)
def process_payment(self, order_id: str, amount: float, idempotency_key: str):
    """Process a payment with exponential backoff retries and idempotency."""
    from db import is_payment_processed, record_payment
    import payment_gateway

    # Idempotency check
    if is_payment_processed(idempotency_key):
        logger.info("Payment already processed", extra={"idempotency_key": idempotency_key})
        return {"status": "already_processed"}

    try:
        result = payment_gateway.charge(order_id=order_id, amount=amount, key=idempotency_key)
        record_payment(idempotency_key, result)
        return {"status": "success", "transaction_id": result["id"]}

    except payment_gateway.TransientError as exc:
        # Exponential backoff: 2^retry seconds (2, 4, 8, 16, 32)
        countdown = 2 ** self.request.retries
        logger.warning(
            "Payment transient failure, retrying",
            extra={"retry": self.request.retries, "countdown": countdown},
        )
        raise self.retry(exc=exc, countdown=countdown)

    except payment_gateway.PermanentError as exc:
        # Don't retry permanent errors — route to DLQ
        logger.error("Payment permanent failure", extra={"error": str(exc)})
        handle_dead_letter.apply_async(
            args=["process_payment", order_id, str(exc)],
            queue="dead_letter",
        )
        raise  # don't retry

@app.task(queue="dead_letter")
def handle_dead_letter(task_name: str, payload: str, error: str):
    """Log, alert, and store failed messages for manual inspection."""
    logger.critical(
        "Message in DLQ",
        extra={"task": task_name, "payload": payload, "error": error},
    )
    # Alert on-call: pagerduty.trigger(f"DLQ: {task_name} failed: {error}")
    # Store for replay: db.dlq_messages.insert(task_name, payload, error, now())

from celery import chain, group, chord
from tasks import fetch_data, process_chunk, aggregate_results, send_report

# Chain: serial pipeline — each result feeds the next
pipeline = chain(
    fetch_data.s(source="s3://bucket/data.csv"),
    process_chunk.s(chunk_size=1000),
    send_report.s(recipient="team@example.com"),
)
pipeline.delay()

# Group: parallel execution — all tasks run simultaneously
parallel = group(
    process_chunk.s(chunk_id=i)
    for i in range(10)
)
parallel.delay()

# Chord: run a group in parallel, then call a callback with all results
chord(
    group(process_chunk.s(chunk_id=i) for i in range(10)),
    aggregate_results.s(),   # receives list of results from all chunks
).delay()

# ETA / countdown: schedule a task for later
process_payment.apply_async(
    args=["order-99", 49.99, "idem-key-xyz"],
    countdown=300,           # run in 5 minutes
    # eta=datetime(2026, 1, 1, 9, 0),  # or at a specific time
)

# Celery Beat: periodic tasks (like cron)
from celery.schedules import crontab

app.conf.beat_schedule = {
    "nightly-report": {
        "task": "tasks.generate_nightly_report",
        "schedule": crontab(hour=2, minute=0),  # 02:00 daily
    },
    "health-check": {
        "task": "tasks.ping_services",
        "schedule": 60.0,   # every 60 seconds
    },
}

from confluent_kafka import Producer, Consumer, KafkaError
import json
import uuid

# ── Producer ──────────────────────────────────────────────────────
producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "acks": "all",              # wait for all replicas to acknowledge (durability)
    "enable.idempotence": True, # exactly-once producer semantics
})

def delivery_report(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} partition {msg.partition()} offset {msg.offset()}")

def publish_order_event(order: dict) -> None:
    event = {
        "event_id": str(uuid.uuid4()),
        "event_type": "order.placed",
        "data": order,
    }
    producer.produce(
        topic="order-events",
        key=str(order["user_id"]),          # partition by user_id for ordering
        value=json.dumps(event).encode(),
        callback=delivery_report,
    )
    producer.poll(0)  # trigger delivery callbacks

publish_order_event({"order_id": "ORD-1234", "user_id": 42, "total": 99.99})
producer.flush()

# ── Consumer ──────────────────────────────────────────────────────
consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "payment-service",         # consumer group
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,           # manual commit for at-least-once safety
})

consumer.subscribe(["order-events"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise KafkaError(msg.error())

        event = json.loads(msg.value().decode())
        try:
            process_order_event(event)             # your handler
            consumer.commit(asynchronous=False)    # commit after successful processing
        except Exception as e:
            print(f"Processing failed: {e}")
            # Do NOT commit — broker will redeliver on restart
finally:
    consumer.close()

from sqlalchemy.orm import Session
from datetime import datetime
import json
import uuid

# ── Model ──────────────────────────────────────────────────────────
class OutboxMessage(Base):
    __tablename__ = "outbox"
    id           = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
    topic        = Column(String, nullable=False)
    payload      = Column(Text, nullable=False)
    created_at   = Column(DateTime, default=datetime.utcnow)
    published_at = Column(DateTime, nullable=True)   # None = not yet published

# ── Write: atomic business data + outbox row ───────────────────────
def place_order(session: Session, order_data: dict) -> Order:
    order = Order(**order_data)
    session.add(order)

    # Write outbox row in the SAME transaction — guaranteed atomicity
    outbox_msg = OutboxMessage(
        topic="order-events",
        payload=json.dumps({
            "event_id": str(uuid.uuid4()),
            "event_type": "order.placed",
            "order_id": str(order.id),
            "user_id": order.user_id,
            "total": float(order.total),
        }),
    )
    session.add(outbox_msg)
    session.commit()    # both order AND outbox row committed atomically
    return order

# ── Relay: poll outbox, publish to Kafka, mark published ───────────
import time

def outbox_relay(session: Session, producer):
    """Run in a background thread or separate process."""
    while True:
        unpublished = (
            session.query(OutboxMessage)
            .filter(OutboxMessage.published_at.is_(None))
            .order_by(OutboxMessage.created_at)
            .limit(100)
            .all()
        )

        for msg in unpublished:
            payload = json.loads(msg.payload)
            producer.produce(
                topic=msg.topic,
                key=payload.get("order_id", ""),
                value=json.dumps(payload).encode(),
            )
            producer.flush()
            msg.published_at = datetime.utcnow()

        session.commit()
        time.sleep(0.5)   # poll every 500 ms; use CDC (Debezium) for lower latency