graph TD
A[Presentation Layer
FastAPI/Django/Flask] --> B[Application Layer
Use Cases / Services]
B --> C[Domain Layer
Entities / Value Objects]
B --> D[Infrastructure Layer
DB / External APIs / Cache]
D --> C
style A fill:#4A90D9,color:#fff
style B fill:#7B68EE,color:#fff
style C fill:#2ECC71,color:#fff
style D fill:#E74C3C,color:#fff
Dependency Rule: Inner layers never import from outer layers. Domain has zero dependencies.
# domain/entities.py — Innermost layer, no external imports
from dataclasses import dataclass, field
from uuid import UUID, uuid4
@dataclass(frozen=True)
class OrderId:
value: UUID = field(default_factory=uuid4)
@dataclass
class Order:
id: OrderId = field(default_factory=OrderId)
items: list[str] = field(default_factory=list)
total: float = 0.0
def add_item(self, item: str, price: float) -> None:
self.items.append(item)
self.total += price
# S — Single Responsibility
class OrderRepository:
"""Only handles persistence of orders."""
def get_by_id(self, order_id: OrderId) -> Order: ...
def save(self, order: Order) -> None: ...
O — Open/Closed: Use Protocol for extension without modification
from typing import Protocol
class PaymentProcessor(Protocol):
def charge(self, amount: float, token: str) -> bool: ...
class StripeProcessor:
def charge(self, amount: float, token: str) -> bool:
# Stripe-specific logic
return True
L — Liskov Substitution: Subtypes must be substitutable
I — Interface Segregation: Fine-grained protocols
class Readable(Protocol):
def read(self) -> str: ...
class Writable(Protocol):
def write(self, data: str) -> None: ...
D — Dependency Inversion: Depend on abstractions
class OrderService:
def __init__(self, repo: OrderRepository, processor: PaymentProcessor):
self._repo = repo
self._processor = processor
sequenceDiagram
participant Client
participant Router as Strategy (Router)
participant UseCase as Use Case
participant Repo as Repository
participant DB
Client->>Router: handle(request)
Router->>Router: select_strategy()
Router->>UseCase: execute(command)
UseCase->>Repo: find_by_id(id)
Repo->>DB: SELECT * FROM orders
DB-->>Repo: rows
Repo-->>UseCase: Order entity
UseCase-->>Router: response
Router-->>Client: HTTP 200
from typing import Any
class NotificationFactory:
"""Factory for creating notification senders."""
_registry: dict[str, type] = {}
@classmethod
def register(cls, channel: str):
def decorator(sender_cls: type):
cls._registry[channel] = sender_cls
return sender_cls
return decorator
@classmethod
def create(cls, channel: str, **kwargs: Any) -> "NotificationSender":
if channel not in cls._registry:
raise ValueError(f"Unknown channel: {channel}")
return cls._registrychannel
@NotificationFactory.register("email")
class EmailSender:
def __init__(self, smtp_host: str = "localhost", **_: Any):
self._host = smtp_host
def send(self, message: str) -> bool:
# Email sending logic
return True
Usage: sender = NotificationFactory.create("email", smtp_host="mail.example.com")
import threading
class DatabaseConnection:
_instance: "DatabaseConnection | None" = None
_lock = threading.Lock()
def __new__(cls, args: Any, *kwargs: Any) -> "DatabaseConnection":
if cls._instance is None:
with cls._lock:
if cls._instance is None: # Double-check locking
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self, dsn: str = "") -> None:
if self._initialized:
return
self._dsn = dsn
self._pool = self._create_pool()
self._initialized = True
from typing import Protocol
class SortStrategy(Protocol):
def sort(self, data: list[int]) -> list[int]: ...
class QuickSort:
def sort(self, data: list[int]) -> list[int]:
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
mid = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return self.sort(left) + mid + self.sort(right)
class TimSort: # Python's built-in
def sort(self, data: list[int]) -> list[int]:
return sorted(data)
class Sorter:
def __init__(self, strategy: SortStrategy) -> None:
self._strategy = strategy
def execute(self, data: list[int]) -> list[int]:
return self._strategy.sort(data)
from typing import Callable
from dataclasses import dataclass, field
@dataclass
class EventBus:
_subscribers: dict[str, list[Callable]] = field(default_factory=dict)
def subscribe(self, event: str, handler: Callable) -> None:
self._subscribers.setdefault(event, []).append(handler)
def publish(self, event: str, data: Any = None) -> None:
for handler in self._subscribers.get(event, []):
handler(data)
Usage
bus = EventBus()
bus.subscribe("order.created", lambda order: print(f"New order: {order.id}"))
bus.publish("order.created", Order())
from abc import ABC, abstractmethod
class AbstractOrderRepository(ABC):
@abstractmethod
def get_by_id(self, order_id: OrderId) -> Order: ...
@abstractmethod
def save(self, order: Order) -> None: ...
@abstractmethod
def find_by_status(self, status: str) -> list[Order]: ...
class SqlOrderRepository(AbstractOrderRepository):
def __init__(self, session: "Session") -> None:
self._session = session
def get_by_id(self, order_id: OrderId) -> Order:
row = self._session.execute(
"SELECT * FROM orders WHERE id = %s", (str(order_id.value),)
).fetchone()
if not row:
raise OrderNotFoundError(order_id)
return Order(id=OrderId(row.id), items=row.items, total=row.total)
def save(self, order: Order) -> None:
self._session.execute(
"INSERT INTO orders (id, items, total) VALUES (%s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET items=%s, total=%s",
(str(order.id.value), order.items, order.total, order.items, order.total),
)
from typing import TypeAlias, Literal, TypedDict
from dataclasses import dataclass, field
UserId: TypeAlias = int
Status = Literal["pending", "active", "closed"]
class UserPayload(TypedDict):
id: UserId
name: str
status: Status
@dataclass(slots=True, kw_only=True)
class AppConfig:
db_dsn: str
redis_url: str = "redis://localhost:6379"
max_workers: int = 4
debug: bool = False
import asyncio
from typing import AsyncIterator
async def fetch_user(user_id: int) -> dict[str, Any]:
async with aiohttp.ClientSession() as session:
async with session.get(f"/api/users/{user_id}") as resp:
resp.raise_for_status()
return await resp.json()
async def stream_results(query: str) -> AsyncIterator[dict]:
"""Async generator for streaming DB results."""
async with asyncpg.connect(DSN) as conn:
async for record in conn.cursor(query):
yield dict(record)
async def process_concurrent(user_ids: list[int]) -> list[dict]:
"""Fan-out concurrent requests with bounded concurrency."""
semaphore = asyncio.Semaphore(10)
async def bounded_fetch(uid: int) -> dict:
async with semaphore:
return await fetch_user(uid)
return await asyncio.gather(*[bounded_fetch(uid) for uid in user_ids])
# Dict comprehension with conditional
user_map: dict[int, str] = {u.id: u.name for u in users if u.is_active}
Generator for memory-efficient pipelines
def process_large_file(path: str) -> Iterator[dict]:
"""Yields parsed records one at a time — constant memory."""
with open(path) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
yield json.loads(line)
Chained generator pipeline
def filter_active(records: Iterator[dict]) -> Iterator[dict]:
yield from (r for r in records if r.get("active"))
def enrich(records: Iterator[dict]) -> Iterator[dict]:
for r in records:
r["enriched"] = True
yield r
# VULNERABLE — NEVER do this
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")
SAFE — Parameterized queries
cursor.execute("SELECT * FROM users WHERE name = %s", (name,))
SAFE — SQLAlchemy with parameter binding
stmt = select(User).where(User.name == bindparam("name"))
result = session.execute(stmt, {"name": name})
from pydantic import BaseModel, EmailStr, field_validator, constr
class SignupRequest(BaseModel):
username: constr(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
email: EmailStr
age: int
@field_validator("age")
@classmethod
def validate_age(cls, v: int) -> int:
if not 13 <= v <= 120:
raise ValueError("Age must be between 13 and 120")
return v
Usage: validates + sanitizes on construction
try:
user = SignupRequest(username="alice123", email="alice@example.com", age=25)
except ValidationError as e:
logger.warning("Invalid signup input: %s", e.json())
import pickle # DANGEROUS for untrusted data
VULNERABLE — arbitrary code execution
data = pickle.loads(untrusted_bytes) # NEVER do this
SAFE — Use JSON or msgpack for untrusted data
import json, msgpack
data = json.loads(untrusted_str) # Safe: data only
data = msgpack.unpackb(untrusted_bytes) # Safe: data only
If you MUST use pickle, restrict with custom Finders
import restrictedpickle # third-party sandbox
import os
import secrets
NEVER hardcode secrets
API_KEY = "sk-live-abc123" # WRONG
ALWAYS use environment variables or secret managers
API_KEY = os.environ["API_KEY"] # Correct
Generate secure tokens
token = secrets.token_urlsafe(32)
password = secrets.compare_digest(provided_hash, stored_hash)
Use .env files for local dev (never commit)
from dotenv import load_dotenv
load_dotenv() # loads .env into os.environ
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["app.example.com", "api.example.com"],
)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://app.example.com"],
allow_methods=["GET", "POST"],
allow_headers=["Authorization"],
)
graph LR
A[main.py] --> B[routers/]
B --> B1[users.py]
B --> B2[orders.py]
A --> C[services/]
C --> C1[user_service.py]
C --> C2[order_service.py]
A --> D[models/]
D --> D1[schemas.py]
D --> D2[domain.py]
A --> E[core/]
E --> E1[config.py]
E --> E2[security.py]
E --> E3[dependencies.py]
style A fill:#4A90D9,color:#fff
style B fill:#7B68EE,color:#fff
style C fill:#2ECC71,color:#fff
style D fill:#F39C12,color:#fff
style E fill:#E74C3C,color:#fff
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
app = FastAPI(title="Order Service", version="1.0.0")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
user = await verify_token(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
return user
@app.post("/orders", status_code=status.HTTP_201_CREATED)
async def create_order(
payload: CreateOrderRequest,
user: User = Depends(get_current_user),
service: OrderService = Depends(get_order_service),
) -> OrderResponse:
try:
order = await service.create_order(user.id, payload)
except InsufficientStockError as e:
raise HTTPException(status_code=409, detail=str(e))
return OrderResponse.from_entity(order)
# views.py — thin controller
class OrderCreateView(APIView):
def post(self, request: Request) -> Response:
cmd = CreateOrderCommand(**request.data)
order = OrderService(repo=SqlOrderRepository()).handle(cmd)
return Response(OrderSerializer(order).data, status=201)
from flask import Flask, Blueprint
def create_app(config_name: str = "production") -> Flask:
app = Flask(__name__)
app.config.from_object(f"config.{config_name.capitalize()}Config")
app.register_blueprint(api_bp, url_prefix="/api/v1")
return app
api_bp = Blueprint("api", __name__)
@api_bp.route("/health")
def health() -> tuple[dict, int]:
return {"status": "ok"}, 200
import pytest
from unittest.mock import AsyncMock, patch
@pytest.fixture
def order_repo() -> AbstractOrderRepository:
return InMemoryOrderRepository()
@pytest.fixture
def mock_payment() -> PaymentProcessor:
processor = AsyncMock(spec=PaymentProcessor)
processor.charge.return_value = True
return processor
@pytest.fixture
def order_service(order_repo, mock_payment) -> OrderService:
return OrderService(repo=order_repo, processor=mock_payment)
@pytest.mark.asyncio
async def test_create_order_success(order_service: OrderService):
cmd = CreateOrderCommand(user_id=1, items=["widget"], total=9.99)
result = await order_service.handle(cmd)
assert result.total == 9.99
assert result.items == ["widget"]
@pytest.mark.asyncio
async def test_create_order_insufficient_stock(order_service: OrderService):
order_service._processor.charge.return_value = False
cmd = CreateOrderCommand(user_id=1, items=["out-of-stock"], total=99.99)
with pytest.raises(PaymentFailedError):
await order_service.handle(cmd)
Coverage: pytest --cov=src --cov-report=html --cov-fail-under=90
# .github/workflows/ci.yml
name: CI
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with: { python-version: "3.12" }
- run: pip install -e ".[dev]"
- run: pytest --cov -x --cov-fail-under=90
- run: ruff check src/
- run: mypy src/
import asyncio
from datetime import datetime, timedelta
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
) -> None:
self._failure_count = 0
self._failure_threshold = failure_threshold
self._recovery_timeout = recovery_timeout
self._last_failure: datetime | None = None
self._state: Literal["closed", "open", "half_open"] = "closed"
async def call(self, fn: Callable, args: Any, *kwargs: Any) -> Any:
if self._state == "open":
if datetime.utcnow() - self._last_failure > timedelta(seconds=self._recovery_timeout):
self._state = "half_open"
else:
raise CircuitOpenError("Circuit breaker is open")
try:
result = await fn(args, *kwargs)
self._on_success()
return result
except Exception:
self._on_failure()
raise
def _on_success(self) -> None:
self._failure_count = 0
self._state = "closed"
def _on_failure(self) -> None:
self._failure_count += 1
self._last_failure = datetime.utcnow()
if self._failure_count >= self._failure_threshold:
self._state = "open"
Retry decorator with exponential backoff
def retry(max_retries: int = 3, base_delay: float = 1.0):
def decorator(fn: Callable) -> Callable:
async def wrapper(args: Any, *kwargs: Any) -> Any:
for attempt in range(max_retries + 1):
try:
return await fn(args, *kwargs)
except (ConnectionError, TimeoutError) as e:
if attempt == max_retries:
raise
delay = base_delay (2 * attempt)
logger.warning("Retry %d/%d after %.1fs: %s", attempt+1, max_retries, delay, e)
await asyncio.sleep(delay)
return wrapper
return decorator
from enum import IntEnum
class ErrorCode(IntEnum):
# 1xxx — Validation
INVALID_INPUT = 1001
MISSING_FIELD = 1002
# 2xxx — Business Logic
INSUFFICIENT_STOCK = 2001
ORDER_NOT_FOUND = 2002
PAYMENT_FAILED = 2003
# 3xxx — Infrastructure
DB_UNAVAILABLE = 3001
UPSTREAM_TIMEOUT = 3002
RATE_LIMITED = 3003
class AppError(Exception):
def __init__(self, code: ErrorCode, message: str, details: dict | None = None):
self.code = code
self.message = message
self.details = details or {}
super().__init__(message)
Global exception handler (FastAPI)
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
logger.error("AppError %d: %s", exc.code, exc.message, extra=exc.details)
status_map = {
ErrorCode.INVALID_INPUT: 400, ErrorCode.MISSING_FIELD: 400,
ErrorCode.INSUFFICIENT_STOCK: 409, ErrorCode.ORDER_NOT_FOUND: 404,
ErrorCode.PAYMENT_FAILED: 402,
ErrorCode.DB_UNAVAILABLE: 503, ErrorCode.UPSTREAM_TIMEOUT: 504,
ErrorCode.RATE_LIMITED: 429,
}
return JSONResponse(
status_code=status_map.get(exc.code, 500),
content={"error": {"code": exc.code, "message": exc.message}},
)
import cProfile
import pstats
Profile a function
cProfile.run("process_large_dataset(data)", sort="cumulative")
Async connection pooling
import asyncpg
pool = await asyncpg.create_pool(
dsn=DSN, min_size=5, max_size=20, max_inactive_connection_lifetime=300,
)
Caching with TTL
from functools import lru_cache
@lru_cache(maxsize=1024)
def get_config(key: str) -> str:
# Expensive lookup cached for process lifetime
return db_lookup(key)
Redis caching with TTL
async def cached_get(key: str, ttl: int = 60) -> dict | None:
cached = await redis.get(key)
if cached:
return json.loads(cached)
data = await expensive_db_query(key)
await redis.setex(key, ttl, json.dumps(data))
return data
# locustfile.py
from locust import HttpUser, task, between
class ApiUser(HttpUser):
wait_time = between(1, 3)
@task
def get_orders(self):
self.client.get("/api/v1/orders", headers={"Authorization": f"Bearer {self.token}"})
Run: locust -f locustfile.py --host=http://localhost:8000 -u 100 -r 10
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY src/ src/
EXPOSE 8000
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
import structlog
logger = structlog.get_logger()
Configured in app startup
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
],
)
Usage — structured, searchable logs
logger.info("order_created", order_id=str(order.id), total=order.total, user_id=user.id)
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
db_dsn: str
redis_url: str = "redis://localhost:6379"
jwt_secret: str
log_level: str = "INFO"
workers: int = 4
model_config = {"env_file": ".env", "env_prefix": "APP_"}
settings = Settings() # Validates on import, fails fast if missing
@app.get("/health")
async def health() -> dict:
checks = {
"db": await check_db(),
"redis": await check_redis(),
}
status = "healthy" if all(checks.values()) else "degraded"
return {"status": status, "checks": checks}
@app.get("/ready")
async def readiness() -> dict:
if not await check_db():
raise HTTPException(503, "Database unavailable")
return {"ready": True}
# Enhanced with rich examples
class OrderResponse(BaseModel):
id: UUID
items: list[str]
total: float
status: Status
model_config = {
"json_schema_extra": {
"examples": [{"id": "a1b2c3", "items": ["widget"], "total": 9.99, "status": "pending"}]
}
}
# ADR-001: Use asyncpg over SQLAlchemy Core for hot paths
Status: Accepted
Context: Profile showed 40% time in SQLAlchemy query building for read-heavy endpoints.
Decision: Use asyncpg directly for the 5 hot-path queries; keep SQLAlchemy for complex writes.
Consequences: Faster reads (+35%), but raw SQL to maintain for those paths.
pip install -e ".[dev]" → pre-commit install.env.example to .env, fill required valuesdocker compose up -d for local Postgres + Redispytest — all green before you startdocs/adr/ for architectural contextKey Principles: Prefer composition over inheritance. Validate at boundaries. Fail fast with clear errors. Keep I/O at the edges. Measure before optimizing. Automate everything.