Advanced

from dataclasses import dataclass
from typing import Protocol, Optional
import uuid

# => Domain value objects — immutable identifiers used across service calls
@dataclass(frozen=True)
class OrderId:
    value: str = ""

    @staticmethod
    def new() -> "OrderId":
        return OrderId(value=str(uuid.uuid4()))  # => Generates UUID-backed stable id
        # => Immutable after creation — cannot be changed by accident

@dataclass(frozen=True)
class ProductId:
    value: str = ""  # => Ties inventory SKU to an order line

# => Protocol defines the Orders service contract — implementable by real service or stub
class OrdersService(Protocol):
    def place_order(self, product_id: ProductId, qty: int) -> OrderId: ...
    # => Returns OrderId so callers need not know internal order structure
    def cancel_order(self, order_id: OrderId) -> bool: ...

# => Inventory service protocol — independent of Orders
class InventoryService(Protocol):
    def reserve(self, product_id: ProductId, qty: int) -> bool: ...
    # => Returns True when stock was successfully reserved
    # => Returns False (not exception) for insufficient stock — expected business outcome
    def release(self, product_id: ProductId, qty: int) -> None: ...

# => Concrete in-process implementation used in tests and local development
class InMemoryOrdersService:
    def __init__(self) -> None:
        self._orders: dict[str, dict] = {}  # => order_id -> order dict

    def place_order(self, product_id: ProductId, qty: int) -> OrderId:
        oid = OrderId.new()  # => Generates fresh id per call
        self._orders[oid.value] = {"product_id": product_id.value, "qty": qty, "status": "PLACED"}
        # => Stores minimal order state; real impl persists to DB
        return oid  # => Returns id so caller can reference the order later

    def cancel_order(self, order_id: OrderId) -> bool:
        order = self._orders.get(order_id.value)
        if order is None:
            return False  # => Order not found — idempotent cancel is fine
        order["status"] = "CANCELLED"  # => Mutates in-place; real impl issues UPDATE
        return True

# => Usage: each service is swapped independently without touching the other
svc = InMemoryOrdersService()
pid = ProductId(value="SKU-001")
oid = svc.place_order(pid, qty=3)  # => Returns OrderId("some-uuid")
print(oid.value)                    # => Output: <uuid string>
print(svc.cancel_order(oid))        # => Output: True

from typing import Callable, Any

# => Route registry maps URL prefixes to handler callables
RouteHandler = Callable[[str, dict], dict]

class StranglerProxy:
    def __init__(self) -> None:
        # => new_routes: paths already migrated to new services
        self._new_routes: dict[str, RouteHandler] = {}
        # => legacy_handler: single catch-all for unmigrated paths
        self._legacy_handler: Optional[RouteHandler] = None

    def register_new(self, prefix: str, handler: RouteHandler) -> None:
        self._new_routes[prefix] = handler
        # => Called once per migrated service; order of registration does not matter

    def set_legacy(self, handler: RouteHandler) -> None:
        self._legacy_handler = handler  # => Legacy system becomes the fallback

    def route(self, path: str, payload: dict) -> dict:
        for prefix, handler in self._new_routes.items():
            if path.startswith(prefix):
                return handler(path, payload)  # => New service handles this path
                # => Return immediately — no fallthrough to legacy
        # => No migrated handler found; delegate to legacy monolith
        if self._legacy_handler:
            return self._legacy_handler(path, payload)  # => Legacy still serves uncharted paths
        raise ValueError(f"No handler for path: {path}")  # => Should not happen in production

# => Simulated new Orders service (already migrated)
def new_orders_handler(path: str, payload: dict) -> dict:
    return {"source": "new_service", "path": path, "data": payload}
    # => New service responds; monolith not involved

# => Simulated legacy monolith (catch-all)
def legacy_handler(path: str, payload: dict) -> dict:
    return {"source": "legacy_monolith", "path": path, "data": payload}

proxy = StranglerProxy()
proxy.set_legacy(legacy_handler)
proxy.register_new("/api/orders", new_orders_handler)  # => Orders migrated

print(proxy.route("/api/orders/123", {"action": "get"}))
# => Output: {'source': 'new_service', 'path': '/api/orders/123', 'data': {'action': 'get'}}
print(proxy.route("/api/products/abc", {"action": "list"}))
# => Output: {'source': 'legacy_monolith', 'path': '/api/products/abc', 'data': {'action': 'list'}}

from dataclasses import dataclass, field
from typing import List, Tuple

# => Step encapsulates a forward action and its compensating action
@dataclass
class SagaStep:
    name: str
    execute: Callable[[], bool]    # => Returns True on success, False on failure
    compensate: Callable[[], None]  # => Undoes the execute action

class SagaOrchestrator:
    def __init__(self, steps: List[SagaStep]) -> None:
        self._steps = steps  # => Steps executed in list order

    def run(self) -> bool:
        completed: List[SagaStep] = []  # => Tracks successful steps for compensation
        for step in self._steps:
            success = step.execute()  # => Drive the participant
            if success:
                completed.append(step)  # => Remember for rollback if later step fails
            else:
                # => Step failed — compensate in reverse order (LIFO)
                for done_step in reversed(completed):
                    done_step.compensate()  # => Each compensation is idempotent by design
                return False  # => Saga failed; system is back to consistent state
        return True  # => All steps succeeded; saga complete

# => Simulated participants with in-memory state
_stock_reserved = False
_payment_charged = False

def reserve_stock() -> bool:
    global _stock_reserved
    _stock_reserved = True  # => Reserve 1 unit of SKU-001
    print("Stock reserved")  # => Output: Stock reserved
    return True

def release_stock() -> None:
    global _stock_reserved
    _stock_reserved = False  # => Compensation: return reserved unit to inventory
    print("Stock released (compensation)")  # => Output: Stock released (compensation)

def charge_payment() -> bool:
    print("Payment failed")   # => Output: Payment failed
    return False               # => Simulate payment processor rejection

def refund_payment() -> None:
    print("Payment refunded (compensation)")  # => Would issue refund if charge had succeeded

steps = [
    SagaStep("reserve", reserve_stock, release_stock),
    SagaStep("payment", charge_payment, refund_payment),
]
result = SagaOrchestrator(steps).run()
print(f"Saga succeeded: {result}")  # => Output: Saga succeeded: False
print(f"Stock released: {not _stock_reserved}")  # => Output: Stock released: True

from collections import defaultdict
from typing import Dict, List

# => Minimal in-process event bus — models a message broker (Kafka, RabbitMQ)
class EventBus:
    def __init__(self) -> None:
        self._handlers: Dict[str, List[Callable]] = defaultdict(list)
        # => Maps event_type -> list of subscriber callables

    def subscribe(self, event_type: str, handler: Callable) -> None:
        self._handlers[event_type].append(handler)  # => Register listener

    def publish(self, event_type: str, payload: dict) -> None:
        for h in self._handlers[event_type]:
            h(payload)  # => Synchronous here; real bus is async (Kafka consumer group)

bus = EventBus()
_stock_held = False  # => Shared mutable state simulating DB

# => Inventory service listens for OrderPlaced and PaymentFailed
def on_order_placed(event: dict) -> None:
    global _stock_held
    _stock_held = True  # => Reserve stock optimistically
    print(f"Inventory: reserved stock for order {event['order_id']}")
    bus.publish("StockReserved", {"order_id": event["order_id"]})
    # => Emits next event; Payment service will react without Inventory knowing

def on_payment_failed(event: dict) -> None:
    global _stock_held
    _stock_held = False  # => Compensation: release stock
    print(f"Inventory: released stock for order {event['order_id']} (compensation)")

# => Payment service listens for StockReserved
def on_stock_reserved(event: dict) -> None:
    print(f"Payment: charging for order {event['order_id']}")
    # => Simulate payment failure
    bus.publish("PaymentFailed", {"order_id": event["order_id"]})
    # => Payment service does NOT call Inventory — it emits an event instead

bus.subscribe("OrderPlaced", on_order_placed)
bus.subscribe("StockReserved", on_stock_reserved)
bus.subscribe("PaymentFailed", on_payment_failed)

bus.publish("OrderPlaced", {"order_id": "ORD-42"})
# => Output: Inventory: reserved stock for order ORD-42
# => Output: Payment: charging for order ORD-42
# => Output: Inventory: released stock for order ORD-42 (compensation)
print(f"Stock held after failure: {_stock_held}")  # => Output: Stock held after failure: False

from http.server import BaseHTTPRequestHandler
import json

# => Route map keyed by (method, path) — path includes version prefix
ROUTES = {
    ("GET", "/v1/users"): lambda: {"version": 1, "users": ["alice"]},
    # => v1 contract: flat list of usernames
    ("GET", "/v2/users"): lambda: {"version": 2, "users": [{"name": "alice", "email": "alice@example.com"}]},
    # => v2 contract: richer objects with email field added
}

def dispatch(method: str, path: str) -> dict:
    handler = ROUTES.get((method, path))
    if handler is None:
        return {"error": "Not found"}  # => 404 equivalent
    return handler()  # => Execute matched route

print(dispatch("GET", "/v1/users"))
# => Output: {'version': 1, 'users': ['alice']}
print(dispatch("GET", "/v2/users"))
# => Output: {'version': 2, 'users': [{'name': 'alice', 'email': 'alice@example.com'}]}

# => Version is negotiated through the Accept header, keeping URLs clean
def dispatch_header(method: str, path: str, accept: str) -> dict:
    # => Parse version from Accept: application/vnd.myapi.v2+json
    version = "v1"  # => Default version if header absent or unversioned
    if "v2" in accept:
        version = "v2"  # => Client explicitly requested v2

    if path == "/users":
        if version == "v2":
            return {"version": 2, "users": [{"name": "alice", "email": "alice@example.com"}]}
        return {"version": 1, "users": ["alice"]}
        # => Same URL, different response shape based on negotiated version
    return {"error": "Not found"}

print(dispatch_header("GET", "/users", "application/json"))
# => Output: {'version': 1, 'users': ['alice']}
print(dispatch_header("GET", "/users", "application/vnd.myapi.v2+json"))
# => Output: {'version': 2, 'users': [{'name': 'alice', 'email': 'alice@example.com'}]}

from dataclasses import dataclass
from typing import Optional

# => Downstream service responses (shared by all BFFs)
def get_user_profile(user_id: str) -> dict:
    return {"id": user_id, "name": "Alice", "email": "alice@example.com", "preferences": {"theme": "dark"}}
    # => Full profile — downstream owns all fields

def get_user_orders(user_id: str) -> list:
    return [{"id": "ORD-1", "total": 99.99, "status": "shipped"}, {"id": "ORD-2", "total": 14.50, "status": "pending"}]
    # => Full order list — may be large

# => Mobile BFF: strips unnecessary fields to reduce bandwidth for cellular connections
def mobile_bff_get_dashboard(user_id: str) -> dict:
    profile = get_user_profile(user_id)
    orders = get_user_orders(user_id)
    return {
        "name": profile["name"],           # => Only name; no email or preferences
        "pending_orders": sum(1 for o in orders if o["status"] == "pending"),
        # => Aggregates count server-side — mobile displays one number, not a list
    }

# => Web BFF: returns richer aggregate with full order list and preferences
def web_bff_get_dashboard(user_id: str) -> dict:
    profile = get_user_profile(user_id)
    orders = get_user_orders(user_id)
    return {
        "profile": profile,           # => Full profile including email and preferences
        "orders": orders,             # => Full order objects — web renders a table
        "order_count": len(orders),   # => Pre-computed convenience field
    }

print(mobile_bff_get_dashboard("u1"))
# => Output: {'name': 'Alice', 'pending_orders': 1}
print(web_bff_get_dashboard("u1"))
# => Output: {'profile': {...}, 'orders': [...], 'order_count': 2}

import time
from enum import Enum, auto

class CBState(Enum):
    CLOSED = auto()    # => Normal operation — calls pass through
    OPEN = auto()      # => Tripped — calls rejected immediately without hitting dependency
    HALF_OPEN = auto() # => Probing — one trial call allowed through

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, probe_timeout: float = 2.0) -> None:
        self._threshold = failure_threshold   # => Trip after this many consecutive failures
        self._probe_timeout = probe_timeout   # => Seconds before attempting probe
        self._state = CBState.CLOSED          # => Starts closed (allowing calls)
        self._failures = 0                    # => Consecutive failure counter
        self._opened_at: Optional[float] = None  # => Timestamp when tripped open

    def call(self, func: Callable[[], str], fallback: Callable[[], str]) -> str:
        if self._state == CBState.OPEN:
            elapsed = time.monotonic() - (self._opened_at or 0)
            if elapsed >= self._probe_timeout:
                self._state = CBState.HALF_OPEN  # => Allow one probe attempt
                print("Circuit: half-open (probing)")
            else:
                return fallback()  # => Still open — return cached/degraded response
                # => Fast-fail: no timeout penalty; dependency gets breathing room

        try:
            result = func()           # => Attempt the real call
            self._on_success()        # => Reset failure count on success
            return result
        except Exception:
            self._on_failure()        # => Increment failure count; may trip breaker
            return fallback()         # => Return degraded response instead of propagating error

    def _on_success(self) -> None:
        self._failures = 0            # => Reset streak on success
        self._state = CBState.CLOSED  # => Close breaker (works for HALF_OPEN probe too)

    def _on_failure(self) -> None:
        self._failures += 1           # => Increment consecutive failure count
        if self._failures >= self._threshold:
            self._state = CBState.OPEN     # => Trip breaker open
            self._opened_at = time.monotonic()  # => Record when breaker opened for timeout
            print(f"Circuit: tripped OPEN after {self._failures} failures")

# => Simulate a flaky downstream service
_call_count = 0

def flaky_service() -> str:
    global _call_count
    _call_count += 1
    if _call_count <= 3:
        raise ConnectionError("service down")  # => First 3 calls fail
    return "fresh data"  # => Recovers on 4th call

def fallback() -> str:
    return "cached data"  # => Stale but acceptable degraded response

cb = CircuitBreaker(failure_threshold=3, probe_timeout=0.0)
for i in range(6):
    result = cb.call(flaky_service, fallback)
    print(f"Call {i+1}: {result}")
# => Output: Call 1: cached data  (failure 1)
# => Output: Call 2: cached data  (failure 2)
# => Output: Circuit: tripped OPEN after 3 failures
# => Output: Call 3: cached data  (failure 3, tripped)
# => Output: Circuit: half-open (probing)
# => Output: Call 4: fresh data   (probe succeeds, breaker closes)
# => Output: Call 5: fresh data
# => Output: Call 6: fresh data

import threading
from contextlib import contextmanager

class Bulkhead:
    def __init__(self, name: str, max_concurrent: int) -> None:
        self._name = name
        self._semaphore = threading.Semaphore(max_concurrent)
        # => Semaphore limits simultaneous calls to max_concurrent
        self._rejected = 0  # => Count of calls rejected due to full pool

    @contextmanager
    def acquire(self):
        acquired = self._semaphore.acquire(blocking=False)
        # => Non-blocking: returns False immediately if pool full (no waiting)
        if not acquired:
            self._rejected += 1
            raise RuntimeError(f"Bulkhead '{self._name}' full — call rejected")
            # => Fail fast: caller gets immediate error; dependency not contacted
        try:
            yield  # => Caller executes inside this context; slot is held
        finally:
            self._semaphore.release()  # => Always return slot to pool

    @property
    def rejected_count(self) -> int:
        return self._rejected  # => Expose for metrics dashboard

# => Two separate bulkheads — slow Payments cannot affect Inventory
payments_bulkhead = Bulkhead("payments", max_concurrent=2)
inventory_bulkhead = Bulkhead("inventory", max_concurrent=5)

def call_payment_service(order_id: str) -> str:
    with payments_bulkhead.acquire():
        # => Simulated slow payment call (would be actual I/O in production)
        return f"payment_ok:{order_id}"  # => Returns result under bulkhead protection

def call_inventory_service(sku: str) -> str:
    with inventory_bulkhead.acquire():
        return f"stock_ok:{sku}"  # => Inventory pool is independent of payments pool

# => Demonstrate isolation: fill payments bulkhead, inventory still works
slots = []
try:
    slots.append(payments_bulkhead._semaphore.acquire(blocking=False))  # => Occupy slot 1
    slots.append(payments_bulkhead._semaphore.acquire(blocking=False))  # => Occupy slot 2
    call_payment_service("ORD-1")  # => Should raise: pool full
except RuntimeError as e:
    print(e)  # => Output: Bulkhead 'payments' full — call rejected

print(call_inventory_service("SKU-1"))  # => Output: stock_ok:SKU-1 (unaffected)

for s in slots:
    if s:
        payments_bulkhead._semaphore.release()  # => Clean up test slots

import random
import time

def exponential_backoff(attempt: int, base: float = 0.1, cap: float = 30.0) -> float:
    # => Exponential growth: base * 2^attempt gives 0.1, 0.2, 0.4, 0.8, 1.6 ...
    delay = min(base * (2 ** attempt), cap)
    # => Cap prevents indefinitely long waits (e.g., never more than 30s)
    jitter = delay * random.uniform(0, 0.5)
    # => Add up to 50% random jitter so concurrent clients do not all retry at T+0.4s
    return delay + jitter  # => Final delay varies per client even for same attempt number

def retry(func: Callable[[], str], max_attempts: int = 5) -> Optional[str]:
    last_error: Optional[Exception] = None
    for attempt in range(max_attempts):
        try:
            return func()  # => Attempt the operation
        except Exception as e:
            last_error = e
            if attempt == max_attempts - 1:
                break  # => Last attempt; do not sleep before giving up
            delay = exponential_backoff(attempt)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s")
            time.sleep(delay)  # => Wait before next attempt; duration increases each time
    raise RuntimeError(f"All {max_attempts} attempts failed: {last_error}") from last_error

# => Simulate service that fails twice then succeeds
_attempt_count = 0

def unstable_call() -> str:
    global _attempt_count
    _attempt_count += 1
    if _attempt_count < 3:
        raise ConnectionError(f"timeout on attempt {_attempt_count}")
        # => First two calls fail with transient error
    return "success"  # => Third call succeeds

# => Demonstrate: retry recovers from transient failures automatically
result = retry(unstable_call, max_attempts=5)
print(f"Result: {result}")  # => Output: Result: success  (after 2 retried failures)

import uuid
import time
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class Span:
    trace_id: str           # => Same for all spans in a single request
    span_id: str            # => Unique per operation
    parent_span_id: Optional[str]  # => Links child to parent in trace tree
    operation: str
    start_time: float = field(default_factory=time.monotonic)
    end_time: Optional[float] = None

    def finish(self) -> None:
        self.end_time = time.monotonic()  # => Records operation duration
        duration_ms = (self.end_time - self.start_time) * 1000
        print(f"[TRACE] trace={self.trace_id} span={self.span_id} parent={self.parent_span_id} "
              f"op={self.operation} duration={duration_ms:.1f}ms")
        # => In production this is sent to Jaeger, Zipkin, or Datadog — not stdout

class Tracer:
    def __init__(self, service_name: str) -> None:
        self._service = service_name

    def start_span(self, operation: str, trace_id: Optional[str] = None,
                   parent_span_id: Optional[str] = None) -> Span:
        return Span(
            trace_id=trace_id or str(uuid.uuid4()),  # => Generate root trace_id if absent
            span_id=str(uuid.uuid4())[:8],            # => New span id per operation
            parent_span_id=parent_span_id,
            operation=f"{self._service}:{operation}",
        )

# => Simulate three services each creating their own spans under the same trace
gateway_tracer = Tracer("api-gateway")
orders_tracer = Tracer("orders-svc")
inventory_tracer = Tracer("inventory-svc")

# => Root span started by API Gateway
root_span = gateway_tracer.start_span("handle_request")
trace_id = root_span.trace_id  # => Propagate this trace_id downstream

# => Orders service creates child span with gateway's trace_id
orders_span = orders_tracer.start_span("place_order", trace_id=trace_id, parent_span_id=root_span.span_id)

# => Inventory service creates grandchild span
inv_span = inventory_tracer.start_span("reserve_stock", trace_id=trace_id, parent_span_id=orders_span.span_id)
inv_span.finish()      # => Inventory finishes first (innermost call)

orders_span.finish()   # => Orders finishes after Inventory returns
root_span.finish()     # => Gateway finishes last
# => All three spans share the same trace_id — reconstruction tool links them into a tree

from typing import Callable
import time

# => SidecarProxy wraps the application handler, adding cross-cutting concerns
class SidecarProxy:
    def __init__(self, app_handler: Callable[[dict], dict]) -> None:
        self._app = app_handler  # => The real application logic — knows nothing about TLS/metrics

    def handle(self, request: dict) -> dict:
        start = time.monotonic()

        # => Sidecar concern 1: enforce mutual TLS certificate validation
        if not request.get("mtls_verified"):
            return {"error": "TLS handshake failed", "status": 401}
            # => App never sees unauthenticated request — sidecar blocks it first

        # => Sidecar concern 2: inject request metadata (trace id, request id)
        request["request_id"] = "req-" + str(int(start * 1000) % 10000)
        # => App can use request_id in logs without implementing header parsing itself

        response = self._app(request)  # => Delegate to application logic

        # => Sidecar concern 3: emit metrics (latency, status code) to Prometheus
        duration_ms = (time.monotonic() - start) * 1000
        print(f"[SIDECAR] method={request.get('method')} path={request.get('path')} "
              f"status={response.get('status', 200)} duration={duration_ms:.1f}ms")
        # => Application code never calls a metrics client; sidecar owns all instrumentation

        return response

# => Application handler: pure business logic, no infrastructure code
def order_app(request: dict) -> dict:
    return {"status": 200, "data": f"Order {request.get('order_id', 'unknown')} retrieved"}
    # => No TLS code, no metrics calls, no log formatting — sidecar owns all of that

proxy = SidecarProxy(order_app)

# => Blocked request (no TLS)
print(proxy.handle({"method": "GET", "path": "/orders/1", "order_id": "1", "mtls_verified": False}))
# => Output: {'error': 'TLS handshake failed', 'status': 401}

# => Allowed request (TLS verified)
print(proxy.handle({"method": "GET", "path": "/orders/1", "order_id": "1", "mtls_verified": True}))
# => Output: [SIDECAR] method=GET path=/orders/1 status=200 duration=...ms
# => Output: {'status': 200, 'data': 'Order 1 retrieved'}

import functools
import time
from typing import Optional

# => Ambassador encapsulates all complexity of calling the Database service
class DatabaseAmbassador:
    def __init__(self, dsn: str, max_retries: int = 3, timeout: float = 5.0) -> None:
        self._dsn = dsn            # => Real ambassador would hold connection pool here
        self._max_retries = max_retries
        self._timeout = timeout
        self._call_count = 0       # => Metrics; ambassador reports to monitoring system

    def query(self, sql: str, params: tuple = ()) -> list:
        # => Ambassador retries transient DB errors transparently — app never sees them
        for attempt in range(self._max_retries):
            try:
                return self._execute(sql, params)  # => Real call through connection pool
            except ConnectionError as e:
                if attempt == self._max_retries - 1:
                    raise  # => Exhausted retries; propagate to application
                wait = 0.1 * (2 ** attempt)  # => Backoff between retries
                time.sleep(wait)
        return []  # => Unreachable; silences type checker

    def _execute(self, sql: str, params: tuple) -> list:
        self._call_count += 1
        # => In production: use psycopg2/asyncpg connection from pool, apply timeout
        if self._call_count == 1:
            raise ConnectionError("transient connection loss")
            # => Simulate flaky first call — ambassador's retry hides this from app
        return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
        # => Returns result as plain list — app sees clean interface

# => Application code calls ambassador; knows nothing about retries, pool, DSN
db = DatabaseAmbassador(dsn="postgresql://localhost/mydb")
rows = db.query("SELECT id, name FROM users WHERE active = %s", (True,))
print(rows)
# => Output: [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
# => (First call failed internally; ambassador retried transparently)
print(f"Total calls made (including retries): {db._call_count}")
# => Output: Total calls made (including retries): 2

from dataclasses import dataclass, field
from typing import List, Any
import json

@dataclass
class DomainEvent:
    event_type: str   # => e.g. "AccountOpened", "MoneyDeposited"
    payload: dict     # => Event data specific to event_type

class EventStore:
    def __init__(self) -> None:
        self._events: List[DomainEvent] = []
        # => Append-only; events are never modified or deleted after appending
        # => In production: stored as rows in a DB with sequence numbers, or in EventStoreDB

    def append(self, event: DomainEvent) -> None:
        self._events.append(event)  # => Append; never update existing events

    def get_events(self) -> List[DomainEvent]:
        return list(self._events)   # => Return copy; callers cannot mutate the store

class BankAccount:
    def __init__(self) -> None:
        self.balance: float = 0.0  # => Derived state; recomputed by replay, never stored directly
        self.account_id: str = ""

    @classmethod
    def replay(cls, events: List[DomainEvent]) -> "BankAccount":
        account = cls()
        for event in events:
            account._apply(event)  # => Apply each event in order to rebuild current state
        return account  # => Returns account in the state it was after all events

    def _apply(self, event: DomainEvent) -> None:
        if event.event_type == "AccountOpened":
            self.account_id = event.payload["account_id"]
            self.balance = 0.0         # => Initial balance always 0 at opening
        elif event.event_type == "MoneyDeposited":
            self.balance += event.payload["amount"]  # => Increase balance by deposit amount
        elif event.event_type == "MoneyWithdrawn":
            self.balance -= event.payload["amount"]  # => Decrease balance by withdrawal amount

# => Simulation: three events recorded over the account's lifetime
store = EventStore()
store.append(DomainEvent("AccountOpened", {"account_id": "ACC-001"}))
store.append(DomainEvent("MoneyDeposited", {"amount": 500.0}))
store.append(DomainEvent("MoneyWithdrawn", {"amount": 200.0}))

# => Replay all events to get current state
account = BankAccount.replay(store.get_events())
print(f"Account: {account.account_id}, Balance: {account.balance}")
# => Output: Account: ACC-001, Balance: 300.0

# => Temporal query: what was the balance after only the first two events?
account_at_t2 = BankAccount.replay(store.get_events()[:2])
print(f"Balance after deposit only: {account_at_t2.balance}")
# => Output: Balance after deposit only: 500.0

from typing import Protocol

# ============================================================
# Module: Orders — owns its own model and service interface
# ============================================================
# => Module boundary enforced by convention: Orders does not import from Billing internals
@dataclass
class Order:
    order_id: str
    customer_id: str
    total: float

class OrderRepository(Protocol):
    def save(self, order: Order) -> None: ...
    def find(self, order_id: str) -> Optional[Order]: ...

class OrderService:
    def __init__(self, repo: OrderRepository) -> None:
        self._repo = repo  # => Injected; module owns the interface, not the implementation

    def place(self, customer_id: str, total: float) -> Order:
        order = Order(order_id=str(uuid.uuid4())[:8], customer_id=customer_id, total=total)
        self._repo.save(order)  # => Persists via injected repo
        return order  # => Returns domain object; other modules receive this via service, not direct DB query

# ============================================================
# Module: Billing — separate domain, communicates via Order domain object only
# ============================================================
# => Billing imports Order (a shared kernel value object) but NOT OrderRepository
@dataclass
class Invoice:
    invoice_id: str
    order_id: str
    amount: float
    paid: bool = False

class BillingService:
    def __init__(self) -> None:
        self._invoices: dict[str, Invoice] = {}

    def issue_invoice(self, order: Order) -> Invoice:
        inv = Invoice(invoice_id=str(uuid.uuid4())[:8], order_id=order.order_id, amount=order.total)
        self._invoices[inv.invoice_id] = inv  # => Billing owns its own storage
        return inv  # => Returns Billing's own Invoice domain object, not Orders' model

# ============================================================
# Composition Root — wires modules together at startup
# ============================================================
class InMemoryOrderRepo:
    def __init__(self) -> None:
        self._store: dict[str, Order] = {}

    def save(self, order: Order) -> None:
        self._store[order.order_id] = order  # => In-memory store; prod uses SQLAlchemy

    def find(self, order_id: str) -> Optional[Order]:
        return self._store.get(order_id)  # => Returns None if not found

order_svc = OrderService(InMemoryOrderRepo())
billing_svc = BillingService()

order = order_svc.place("cust-1", total=149.99)  # => Creates and persists order
invoice = billing_svc.issue_invoice(order)         # => Billing receives Order domain object
print(f"Order: {order.order_id}, Invoice: {invoice.invoice_id}, Amount: {invoice.amount}")
# => Output: Order: <id>, Invoice: <id>, Amount: 149.99

from dataclasses import dataclass
from typing import Optional

# ============================================================
# Slice: Place Order — all layers for this feature in one place
# ============================================================
# => Request object: input contract for this slice
@dataclass
class PlaceOrderRequest:
    customer_id: str
    product_id: str
    quantity: int

# => Response object: output contract for this slice
@dataclass
class PlaceOrderResponse:
    order_id: str
    status: str
    total: float

# => Handler: contains all logic for this slice (no shared service layer)
class PlaceOrderHandler:
    def __init__(self, price_per_unit: float = 10.0) -> None:
        self._price = price_per_unit   # => In prod: injected repository + price service
        self._orders: list = []

    def handle(self, request: PlaceOrderRequest) -> PlaceOrderResponse:
        total = self._price * request.quantity  # => Business rule: total = price * qty
        order_id = f"ORD-{len(self._orders) + 1:04d}"  # => Simple sequential id for demo
        self._orders.append({"id": order_id, "customer": request.customer_id, "total": total})
        # => Persists order (would use Unit of Work in production)
        return PlaceOrderResponse(order_id=order_id, status="placed", total=total)
        # => Returns slice-specific response; no generic response envelope needed

# ============================================================
# Slice: Get Order — separate slice, no shared repository dependency
# ============================================================
@dataclass
class GetOrderRequest:
    order_id: str

@dataclass
class GetOrderResponse:
    order_id: str
    total: Optional[float]
    found: bool

class GetOrderHandler:
    def __init__(self, placed_orders: list) -> None:
        self._orders = placed_orders  # => Shared reference to same in-memory list for demo
        # => In prod: separate read-side repository (CQRS read model)

    def handle(self, request: GetOrderRequest) -> GetOrderResponse:
        order = next((o for o in self._orders if o["id"] == request.order_id), None)
        # => Linear scan for demo; prod uses indexed DB query
        if order is None:
            return GetOrderResponse(order_id=request.order_id, total=None, found=False)
        return GetOrderResponse(order_id=order["id"], total=order["total"], found=True)

# => Usage: each slice is invoked independently
place_handler = PlaceOrderHandler(price_per_unit=12.50)
resp = place_handler.handle(PlaceOrderRequest(customer_id="C1", product_id="P1", quantity=4))
print(resp)  # => Output: PlaceOrderResponse(order_id='ORD-0001', status='placed', total=50.0)

get_handler = GetOrderHandler(place_handler._orders)
get_resp = get_handler.handle(GetOrderRequest(order_id="ORD-0001"))
print(get_resp)  # => Output: GetOrderResponse(order_id='ORD-0001', total=50.0, found=True)

# ============================================================
# shared_kernel.py — the agreed-upon shared subset
# ============================================================
# => Only value objects and domain events go in the shared kernel
# => Never: repositories, services, application logic, database schemas
@dataclass(frozen=True)
class Money:
    amount: float
    currency: str  # => ISO 4217 currency code e.g. "USD"

    def __add__(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError(f"Cannot add {self.currency} and {other.currency}")
        return Money(amount=self.amount + other.amount, currency=self.currency)
        # => Returns new Money; immutable — amount and currency cannot change after creation

    def __str__(self) -> str:
        return f"{self.currency} {self.amount:.2f}"  # => e.g. "USD 99.50"

@dataclass(frozen=True)
class OrderId:
    value: str  # => Shared identifier type; both Orders and Billing reference the same type

# ============================================================
# orders_domain.py — uses shared kernel types
# ============================================================
@dataclass
class OrderLine:
    product_id: str
    price: Money   # => Money from shared kernel — no conversion needed between domains
    qty: int

    @property
    def subtotal(self) -> Money:
        return Money(amount=self.price.amount * self.qty, currency=self.price.currency)
        # => Uses shared kernel Money; result is also a shared kernel type

# ============================================================
# billing_domain.py — uses the same shared kernel types independently
# ============================================================
@dataclass
class Invoice:
    order_id: OrderId  # => Shared kernel OrderId — consistent across both domains
    total: Money       # => Same Money type; no translation layer needed between domains

    def is_overdue(self, days_outstanding: int) -> bool:
        return days_outstanding > 30  # => Billing's own rule; not in shared kernel

# => Usage: both domains speak the same Money and OrderId language
line = OrderLine(product_id="P1", price=Money(10.0, "USD"), qty=3)
invoice = Invoice(order_id=OrderId("ORD-42"), total=line.subtotal)
print(f"Invoice total: {invoice.total}")          # => Output: Invoice total: USD 30.00
print(f"Overdue (35 days): {invoice.is_overdue(35)}")  # => Output: Overdue (35 days): True

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TypeVar, Generic

T = TypeVar("T")

class Specification(ABC, Generic[T]):
    @abstractmethod
    def is_satisfied_by(self, candidate: T) -> bool:
        ...  # => Subclasses implement the specific predicate

    def and_(self, other: "Specification[T]") -> "AndSpecification[T]":
        return AndSpecification(self, other)  # => Composes two specs with AND logic

    def or_(self, other: "Specification[T]") -> "OrSpecification[T]":
        return OrSpecification(self, other)   # => Composes two specs with OR logic

    def not_(self) -> "NotSpecification[T]":
        return NotSpecification(self)         # => Negates this specification

class AndSpecification(Specification[T]):
    def __init__(self, a: Specification[T], b: Specification[T]) -> None:
        self._a, self._b = a, b  # => Holds references to both specs

    def is_satisfied_by(self, candidate: T) -> bool:
        return self._a.is_satisfied_by(candidate) and self._b.is_satisfied_by(candidate)
        # => Both specs must be satisfied; short-circuits if first is False

class OrSpecification(Specification[T]):
    def __init__(self, a: Specification[T], b: Specification[T]) -> None:
        self._a, self._b = a, b

    def is_satisfied_by(self, candidate: T) -> bool:
        return self._a.is_satisfied_by(candidate) or self._b.is_satisfied_by(candidate)
        # => Either spec satisfied is sufficient

class NotSpecification(Specification[T]):
    def __init__(self, spec: Specification[T]) -> None:
        self._spec = spec  # => Wraps the spec to negate

    def is_satisfied_by(self, candidate: T) -> bool:
        return not self._spec.is_satisfied_by(candidate)  # => Inverts wrapped spec result

# => Domain: Order eligibility for discount
@dataclass
class Order:
    total: float
    customer_tier: str  # => "gold", "silver", "bronze"
    item_count: int

class HighValueOrder(Specification[Order]):
    def is_satisfied_by(self, o: Order) -> bool:
        return o.total >= 100.0  # => Orders over $100 qualify as high-value

class GoldCustomer(Specification[Order]):
    def is_satisfied_by(self, o: Order) -> bool:
        return o.customer_tier == "gold"  # => Only gold-tier customers

class BulkOrder(Specification[Order]):
    def is_satisfied_by(self, o: Order) -> bool:
        return o.item_count >= 10  # => 10+ items qualify as bulk

# => Business rule: eligible for discount if gold customer OR (high value AND bulk)
discount_eligible = GoldCustomer().or_(HighValueOrder().and_(BulkOrder()))

o1 = Order(total=150.0, customer_tier="gold", item_count=3)
o2 = Order(total=150.0, customer_tier="silver", item_count=12)
o3 = Order(total=50.0, customer_tier="bronze", item_count=5)

print(discount_eligible.is_satisfied_by(o1))  # => Output: True (gold customer)
print(discount_eligible.is_satisfied_by(o2))  # => Output: True (high value AND bulk)
print(discount_eligible.is_satisfied_by(o3))  # => Output: False (none of the criteria met)

from __future__ import annotations
from dataclasses import dataclass
from typing import Optional

@dataclass
class HttpRequest:
    path: str
    api_key: Optional[str]
    caller_id: str

@dataclass
class HttpResponse:
    status: int
    body: str

Handler = Callable[[HttpRequest], HttpResponse]

class MiddlewareChain:
    def __init__(self) -> None:
        self._middlewares: list[Callable[[HttpRequest, Handler], HttpResponse]] = []
        # => Each middleware receives request AND a next callable to forward to

    def use(self, middleware: Callable[[HttpRequest, Handler], HttpResponse]) -> "MiddlewareChain":
        self._middlewares.append(middleware)
        return self  # => Returns self for fluent builder API

    def build(self, final_handler: Handler) -> Handler:
        handler = final_handler  # => Start building chain from the end (innermost handler)
        for mw in reversed(self._middlewares):
            captured = handler  # => Capture in closure to avoid late-binding issue
            handler = lambda req, h=captured, m=mw: m(req, h)
            # => Wrap each middleware around the current chain; innermost runs last
        return handler  # => Returns outermost handler which drives the entire chain

# => Auth middleware: rejects requests without valid API key
VALID_KEYS = {"key-abc", "key-xyz"}  # => Simplified key store; prod uses DB/cache

def auth_middleware(req: HttpRequest, next_handler: Handler) -> HttpResponse:
    if req.api_key not in VALID_KEYS:
        return HttpResponse(status=401, body="Unauthorized")
        # => Chain terminates here; next_handler never called for invalid keys
    return next_handler(req)  # => Valid key: pass to next middleware in chain

# => Rate limit middleware: allows up to 2 calls per caller_id (simplified)
_call_counts: dict[str, int] = {}

def rate_limit_middleware(req: HttpRequest, next_handler: Handler) -> HttpResponse:
    _call_counts[req.caller_id] = _call_counts.get(req.caller_id, 0) + 1
    if _call_counts[req.caller_id] > 2:
        return HttpResponse(status=429, body="Too Many Requests")
        # => Chain terminates; auth already passed but rate limit blocks further
    return next_handler(req)  # => Within limit: continue to business handler

# => Final business handler
def order_handler(req: HttpRequest) -> HttpResponse:
    return HttpResponse(status=200, body=f"Orders for {req.caller_id}")

chain = MiddlewareChain().use(auth_middleware).use(rate_limit_middleware).build(order_handler)

print(chain(HttpRequest("/orders", None, "caller-1")))
# => Output: HttpResponse(status=401, body='Unauthorized')
print(chain(HttpRequest("/orders", "key-abc", "caller-1")))
# => Output: HttpResponse(status=200, body='Orders for caller-1')
print(chain(HttpRequest("/orders", "key-abc", "caller-1")))
# => Output: HttpResponse(status=200, body='Orders for caller-1')
print(chain(HttpRequest("/orders", "key-abc", "caller-1")))
# => Output: HttpResponse(status=429, body='Too Many Requests')

from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

# ============================================================
# Stable domain hierarchy — these classes do NOT change when new ops are added
# ============================================================
class Component(ABC):
    @abstractmethod
    def accept(self, visitor: "ComponentVisitor") -> None: ...

@dataclass
class Service(Component):
    name: str
    replicas: int
    cpu_millicores: int  # => e.g. 500 = 0.5 CPU cores

    def accept(self, visitor: "ComponentVisitor") -> None:
        visitor.visit_service(self)  # => Dispatches to correct visit_* method via double dispatch

@dataclass
class Database(Component):
    name: str
    storage_gb: int
    multi_az: bool  # => Multi-AZ deployment for high availability

    def accept(self, visitor: "ComponentVisitor") -> None:
        visitor.visit_database(self)  # => Database-specific visit method

@dataclass
class Architecture:
    components: List[Component]

    def accept(self, visitor: "ComponentVisitor") -> None:
        visitor.visit_architecture(self)
        for c in self.components:
            c.accept(visitor)  # => Each component dispatches to its own visit method

# ============================================================
# Visitor interface — new operations implement this without touching domain classes
# ============================================================
class ComponentVisitor(ABC):
    @abstractmethod
    def visit_service(self, svc: Service) -> None: ...
    @abstractmethod
    def visit_database(self, db: Database) -> None: ...
    @abstractmethod
    def visit_architecture(self, arch: Architecture) -> None: ...

# ============================================================
# Concrete visitor 1: Cost estimation — new operation, no domain changes
# ============================================================
class CostEstimator(ComponentVisitor):
    def __init__(self) -> None:
        self.total_monthly_usd: float = 0.0

    def visit_architecture(self, arch: Architecture) -> None:
        print(f"Estimating cost for {len(arch.components)} components")

    def visit_service(self, svc: Service) -> None:
        cost = svc.replicas * (svc.cpu_millicores / 1000) * 30 * 0.05
        # => $0.05 per vCPU per hour, 30 days; simplified cloud pricing model
        self.total_monthly_usd += cost
        print(f"  Service {svc.name}: ${cost:.2f}/month")

    def visit_database(self, db: Database) -> None:
        cost = db.storage_gb * 0.10 * (2 if db.multi_az else 1)
        # => $0.10 per GB/month, doubled for multi-AZ redundancy
        self.total_monthly_usd += cost
        print(f"  Database {db.name}: ${cost:.2f}/month")

# ============================================================
# Concrete visitor 2: Compliance check — another new operation, still no domain changes
# ============================================================
class ComplianceChecker(ComponentVisitor):
    def __init__(self) -> None:
        self.violations: List[str] = []

    def visit_architecture(self, arch: Architecture) -> None:
        pass  # => No architecture-level compliance rules in this example

    def visit_service(self, svc: Service) -> None:
        if svc.replicas < 2:
            self.violations.append(f"Service '{svc.name}' has only {svc.replicas} replica (min 2 for HA)")
            # => Single replica violates high-availability requirement

    def visit_database(self, db: Database) -> None:
        if not db.multi_az:
            self.violations.append(f"Database '{db.name}' is not multi-AZ (compliance requirement)")
            # => Single-AZ database violates disaster-recovery policy

arch = Architecture(components=[
    Service("orders-api", replicas=3, cpu_millicores=500),
    Service("worker", replicas=1, cpu_millicores=1000),  # => Only 1 replica — will fail compliance
    Database("orders-db", storage_gb=100, multi_az=True),
    Database("cache-db", storage_gb=20, multi_az=False),  # => Not multi-AZ — will fail compliance
])

cost_visitor = CostEstimator()
arch.accept(cost_visitor)
print(f"Total: ${cost_visitor.total_monthly_usd:.2f}/month")
# => Output: Estimating cost for 4 components
# => Output:   Service orders-api: $2.25/month
# => Output:   Service worker: $1.50/month
# => Output:   Database orders-db: $20.00/month
# => Output:   Database cache-db: $2.00/month
# => Output: Total: $25.75/month

compliance = ComplianceChecker()
arch.accept(compliance)
print(f"Violations: {compliance.violations}")
# => Output: Violations: ["Service 'worker' has only 1 replica...", "Database 'cache-db' is not multi-AZ..."]