Python,
end to end.

a = [1, 2, 3]
b = a                           # one object, two names
b.append(4)
print(a)                        # [1, 2, 3, 4] -- a sees it
print(a is b, id(a) == id(b))   # True True

def rebind(lst):
    lst.append(99)              # mutates the caller's object
    lst = [0]                   # rebinds the LOCAL name only
    return lst

x = [1]
rebind(x)
print(x)                        # [1, 99] -- append survived, rebind did not

def add(item, bucket=[]):        # bucket is created ONCE, at def time
    bucket.append(item)
    return bucket

add("a")            # ["a"]
add("b")            # ["a", "b"]  <- the same list, still there
add.__defaults__    # (["a", "b"],)  the evidence, on the function object

def add(item, bucket=None):      # the None-sentinel fix
    if bucket is None:
        bucket = []              # a fresh list per call
    bucket.append(item)
    return bucket

class Money:
    __slots__ = ("amount", "currency")

    def __init__(self, amount, currency):
        self.amount, self.currency = amount, currency

    def __repr__(self):          # what logs, reprs and tracebacks show
        return f"Money({self.amount!r}, {self.currency!r})"

    def __eq__(self, other):
        if not isinstance(other, Money):
            return NotImplemented    # let the other type try
        return (self.amount, self.currency) == (other.amount, other.currency)

    def __hash__(self):          # define WITH __eq__ or the type is unhashable
        return hash((self.amount, self.currency))

# the container and context-manager protocols -- no base class, just the methods
class Basket:
    def __init__(self):
        self._items: list[str] = []

    def __len__(self) -> int:                 # bool(basket), len(basket)
        return len(self._items)

    def __getitem__(self, i: int) -> str:      # basket[0], and makes it iterable for free
        return self._items[i]

    def __iadd__(self, item: str) -> "Basket": # basket += "eggs"
        self._items.append(item)
        return self

class transaction:                             # with transaction(conn): ...
    def __init__(self, conn):
        self.conn = conn

    def __enter__(self):
        self.conn.execute("BEGIN")
        return self.conn

    def __exit__(self, exc_type, exc, tb) -> bool:
        self.conn.execute("ROLLBACK" if exc_type else "COMMIT")
        return False                           # False: never swallow the exception

def connect(host, /, port=5432, *, timeout=5.0, sslmode="require"):
    ...

connect("db1")                       # host is positional-only
connect("db1", 6432)                 # port may be positional
connect("db1", timeout=1.0)          # timeout MUST be by keyword
connect(host="db1")                  # TypeError: host is positional-only

opts = {"timeout": 1.0, "sslmode": "disable"}
connect("db1", **opts)               # unpack a dict into keyword arguments

args = ("db1", 6432)
connect(*args, timeout=1.0)          # unpack an iterable into positionals

# Materialises every line: peak memory is the whole file.
total = sum([int(line) for line in open("events.log")])

# Streams: one line resident at a time, same answer.
with open("events.log") as f:
    total = sum(int(line) for line in f)

# A pipeline of generators -- still one element in flight.
def parse(lines):
    for line in lines:
        if line.strip():
            yield json.loads(line)

with open("events.ndjson") as f:
    errors = (e for e in parse(f) if e["level"] == "error")
    first_ten = list(itertools.islice(errors, 10))   # stops reading after 10

import functools, time

def timed(fn):
    @functools.wraps(fn)                 # keeps __name__, __doc__, __wrapped__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            ms = (time.perf_counter() - start) * 1000
            log.info("%s took %.1f ms", fn.__name__, ms)
    return wrapper

@timed                                   # charge = timed(charge)
def charge(order_id):
    ...

# the late-binding trap: a closure reads the variable, not a snapshot of it
handlers = []
for route in ["orders", "users", "health"]:
    handlers.append(lambda: log.info("handling %s", route))

[h() for h in handlers]        # "handling health" three times -- route is one shared cell

# the fix: bind it as a default argument, evaluated once, right now
handlers = []
for route in ["orders", "users", "health"]:
    handlers.append(lambda route=route: log.info("handling %s", route))

[h() for h in handlers]        # "handling orders", "handling users", "handling health"

from typing import Literal, Protocol, TypedDict

class Row(TypedDict):          # the shape of JSON you do not own
    id: int
    status: Literal["open", "closed"]

class Store(Protocol):         # structural: no base class, no registration
    def get(self, key: str) -> Row | None: ...

def load(store: Store, key: str) -> Row | None:
    return store.get(key)      # any object with a matching get() type-checks

# PEP 695 (3.12+): generic syntax with no import, T scoped to the function
def first[T](xs: list[T]) -> T:
    return xs[0]

# the older spelling, still what most of the ecosystem reads and writes
from typing import TypeVar, cast
U = TypeVar("U")

def last(xs: list[U]) -> U:
    return xs[-1]

from __future__ import annotations   # hints stop being evaluated eagerly; the default on 3.14+ (PEP 649)

def handler(req: Request) -> Response: ...  # forward reference, unquoted -- lazy annotations resolve it

n = cast(int, config.get("timeout"))  # zero run-time effect -- a lie you are asserting is true

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True)
class User:
    id: int
    name: str
    tags: list[str] = field(default_factory=list)   # not tags: list = []

    def __post_init__(self) -> None:
        if self.id <= 0:                            # frozen: use object.__setattr__
            raise ValueError("id must be positive") # to normalise fields here

u = User(1, "ada")
u.name = "grace"        # FrozenInstanceError -- hashable, safe as a dict key

# NamedTuple — when tuple-ness IS the point: unpack it, hash it, index it
from typing import NamedTuple

class Point(NamedTuple):
    x: float
    y: float

p = Point(1.0, 2.0)
x, y = p                       # unpacks like any tuple
p[0]                           # 1.0 — still indexable
p._replace(x=9.0)              # returns a NEW Point; p itself never changes
{p: "origin-ish"}              # hashable out of the box, no frozen= needed

# TypedDict — for a JSON shape you do not own; a plain dict at run time
from typing import TypedDict

class UserPayload(TypedDict):
    id: int
    name: str
    tags: list[str]

def greet(u: UserPayload) -> str:
    return f"hello {u['name']}"    # dict access, not attribute access

raw: UserPayload = {"id": 1, "name": "ada", "tags": []}
raw["naem"] = "typo"               # mypy/pyright catch this; nothing does at run time
greet({"id": 1})                   # also only a type-checker error — TypedDict never runs

# BaseModel — the same record, but arriving from outside and needing to be believed
from pydantic import BaseModel, Field, ValidationError

class User(BaseModel):
    id: int = Field(gt=0)
    name: str
    tags: list[str] = []           # pydantic copies the default per instance — safe here

try:
    u = User.model_validate({"id": "1", "name": "ada"})  # "1" coerced to 1
except ValidationError as e:
    log.warning("bad user payload: %s", e.errors())

u.model_dump()                     # -> plain dict, JSON-safe with mode="json"

Records make tagged unions, and match (3.10+) is how you consume one — the checker narrows each arm, and assert_never turns “forgot a variant” into a type error instead of a production bug:

# dispatch on a tagged union: match narrows, assert_never proves exhaustiveness
from dataclasses import dataclass
from typing import assert_never

@dataclass
class Created:
    id: int

@dataclass
class Deleted:
    id: int
    reason: str

def apply(ev: Created | Deleted) -> str:
    match ev:
        case Created(id=i):            return f"created {i}"
        case Deleted(id=i, reason=r):  return f"deleted {i}: {r}"
        case _:                        assert_never(ev)  # unreachable -- or the union grew

# what `import payments.charge` actually does, once per process
import sys
sys.path       # searched in order; "" means the current directory, first
sys.modules    # the cache — every later import is just a dict lookup

# payments/config.py — a module body runs exactly once, at first import
print("configuring")        # printed once, however many modules import it
SETTINGS = load_settings()  # module-level state is a singleton by accident

# the guard: this file can be imported as a library AND run as a script
if __name__ == "__main__":
    main()

# circular imports fail halfway through, not at the import line:
#   ImportError: cannot import name 'Order' from partially initialized
#   module 'payments.models' (most likely due to a circular import)
# the fix is a direction, not a trick — move the shared type into its own module

import contextlib

class Transient(Exception): ...    # worth a retry: timeout, 503, connection reset
class Terminal(Exception): ...     # never retry: 400, 403, bad payload

def fetch(url: str) -> bytes:
    try:
        return http.get(url)
    except TimeoutError as err:
        raise Transient(f"timeout: {url}") from err
    except PermissionError as err:
        raise Terminal(f"forbidden: {url}") from err

with contextlib.suppress(FileNotFoundError):
    os.remove(tmp)                 # "already gone is fine" -- not a bare except

# all four clauses, each with a job try/except alone cannot do
def load_batch(path: str) -> list[Row]:
    try:
        fh = open(path)
    except FileNotFoundError:
        return []                  # handled: an empty batch is a valid outcome
    else:
        with fh:                   # only runs if open() did NOT raise; the with
            return parse(fh)       # closes fh even if parse() raises -- next topic
    finally:
        log.info("load_batch done: %s", path)    # always -- even on the return above

import contextlib, logging, time

@contextlib.contextmanager
def timed(label: str):
    start = time.perf_counter()
    try:
        yield                       # the body of the with block runs here
    finally:                        # runs on the exception path too
        ms = (time.perf_counter() - start) * 1000
        logging.info("%s took %.1f ms", label, ms)

with timed("query"), conn.transaction():    # left to right in, right to left out
    rows = conn.fetch_all(sql)

# the same work, three pools — the numbers are the whole argument
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# CPU-bound: threads take turns holding the GIL, so nothing is gained
with ThreadPoolExecutor(8) as ex:
    list(ex.map(tokenise_row, rows))      # 4.1 s on 8 cores -- pure Python

# CPU-bound: separate interpreters, separate GILs, real parallelism
with ProcessPoolExecutor(8) as ex:
    list(ex.map(tokenise_row, rows))      # 0.6 s — and pickling costs

# I/O-bound: the GIL is released around the syscall, so threads are fine
with ThreadPoolExecutor(32) as ex:
    list(ex.map(fetch, urls))             # 32 sockets, one core, no problem

# and from async code, the escape hatch for one blocking call:
row = await asyncio.to_thread(legacy_driver.query, sql)

import asyncio, httpx

async def fetch_all(urls: list[str]) -> list[bytes]:
    sem = asyncio.Semaphore(20)                  # bound the fan-out

    async with httpx.AsyncClient() as client:
        async def one(u: str) -> bytes:
            async with sem:                      # at most 20 in flight
                r = await client.get(u)
                return r.content

        async with asyncio.timeout(30):          # the whole batch, not each call
            return await asyncio.gather(*(one(u) for u in urls))

asyncio.run(fetch_all(urls))

# the same fan-out three ways — choose by what the work is waiting on

# 1. waiting on the network, and the library is async-native
results = await asyncio.gather(*(fetch(u) for u in urls))

# 2. waiting on the network, but the library is blocking and you cannot change it
with ThreadPoolExecutor(32) as ex:
    results = list(ex.map(requests.get, urls))

# 3. not waiting at all — burning CPU
with ProcessPoolExecutor() as ex:
    results = list(ex.map(compress, chunks))

# the start method is a real decision, not a detail:
#   fork   — fast, copy-on-write, UNSAFE in a process that has threads
#   spawn  — a fresh interpreter; everything must be picklable
multiprocessing.set_start_method("spawn")   # default on macOS/Windows only; Linux went
# fork -> forkserver in 3.14 -- set it explicitly so every platform agrees

import sys

sys.getsizeof(1)             # 28   -- refcount, type, size, one digit
sys.getsizeof(2**64)         # 36   -- three 30-bit digits now
sys.getsizeof("hello")       # 54   -- 49 of header, 1 byte per ASCII char
sys.getsizeof([0] * 1000)    # 8056 -- 56 header + 8 per slot, and nothing else

xs = [1234 + i for i in range(1_000_000)]
sys.getsizeof(xs)                    # about 8.4 MB: the pointer array only
sum(sys.getsizeof(x) for x in xs)    # about 28 MB more: what it points at

class Point:                         # a dict per instance, any attribute allowed
    def __init__(self, x: float, y: float) -> None:
        self.x, self.y = x, y

class Slim:
    __slots__ = ("x", "y")           # fixed offsets, no __dict__, typos raise
    def __init__(self, x: float, y: float) -> None:
        self.x, self.y = x, y

# a cycle: refcounting alone never frees this pair -- only the collector can
class Node:
    def __init__(self, name: str) -> None:
        self.name, self.parent, self.children = name, None, []

    def add(self, child: "Node") -> None:
        child.parent = self          # parent -> child -> parent: a cycle
        self.children.append(child)

# fix: the back-reference doesn't need to keep the parent alive
import weakref
child.parent = weakref.ref(root)     # dies with root, not with child

import gc, os
gc.freeze()                          # at start-up, after imports, before fork():
os.fork()                            # stops the collector from touching every object's refcount
                                      # and un-sharing every copy-on-write page across workers

# 1. where does the time go
python -m cProfile -o run.prof -m myapp.batch
python -c "import pstats; pstats.Stats('run.prof').sort_stats('cumulative').print_stats(15)"

# 2. is this really faster than that
python -m timeit -s "xs = list(range(1000))" "sum(xs)"

# 3. what grew between here and there
import tracemalloc
tracemalloc.start()
before = tracemalloc.take_snapshot()
handle_batch()
for stat in tracemalloc.take_snapshot().compare_to(before, "lineno")[:10]:
    print(stat)             # file:line, size delta, count delta

# 4. production, right now, with no restart and no redeploy
py-spy top --pid 4131
py-spy dump --pid 4131      # every thread's Python stack, as it stands

# the rewrites that move the needle, and the one measurement that justifies them
out = "".join(parts)                  # not s += x in a loop — that is quadratic
seen = set(ids)                       # not `x in some_list` — O(n) every check
by_id = {r.id: r for r in rows}       # index once, then look up n times
total = sum(r.amount for r in rows)   # a generator: no intermediate list

# hoisting an attribute lookup out of a hot loop — a real 10-20% on tight loops
append = out.append
for x in data:
    append(transform(x))

# @cache for a pure function called with a small set of arguments
@functools.cache
def tax_rate(country: str, year: int) -> Decimal: ...

# and none of it goes in without this:
#   python -m timeit -s "from mod import f, data" "f(data)"

# pyproject.toml — one file for metadata, dependencies and tool config
[project]
name = "payments"
requires-python = ">=3.13"
dependencies = ["httpx>=0.27,<0.29", "pydantic>=2.7", "sqlalchemy>=2.0"]

[dependency-groups]           # PEP 735: dev deps that are not shipped
dev = ["pytest>=8", "ruff>=0.6", "mypy>=1.11"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

# uv lock          → writes uv.lock, commit it
# uv sync --frozen → installs exactly the lock, fails if it is stale
# uv run pytest    → runs inside the project env, no activation needed

# pyproject.toml — the whole quality config, in the file you already have
[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "RUF", "ASYNC", "S"]
#  E/F pycodestyle+pyflakes · I imports · UP pyupgrade
#  B bugbear · SIM simplify · ASYNC async traps · S security
ignore = ["E501"]             # the formatter owns line length

[tool.ruff.lint.per-file-ignores]
"tests/*" = ["S101"]          # assert is the point of a test

# pyproject.toml — strict everywhere, with named, temporary exceptions
[tool.mypy]
python_version = "3.13"
strict = true
warn_unreachable = true
warn_unused_ignores = true    # deletes the ignores you no longer need

[[tool.mypy.overrides]]       # the legacy corner, listed explicitly
module = ["payments.legacy.*"]
disallow_untyped_defs = false

[[tool.mypy.overrides]]       # a dependency that ships no stubs
module = ["some_untyped_lib.*"]
ignore_missing_imports = true

# pyproject.toml — src layout, one package, one source of truth for the version
[project]
name = "payments"
version = "0.4.1"           # read at run time via importlib.metadata

[tool.hatch.build.targets.wheel]
packages = ["src/payments"]

# and the check that tells you which layout you are really in:
#   $ pip install -e . && python -c "import payments; print(payments.__file__)"
#
#   src layout  → .../site-packages/payments/__init__.py   the installed copy
#   flat layout → ./payments/__init__.py                   the source tree wins
#
# Only the first one is testing what your users will actually install.

# main.py — the entire stack from the diagram, wired once, closed once
from contextlib import asynccontextmanager

engine  = create_async_engine(settings.database_url, pool_size=10, max_overflow=5)
Session = async_sessionmaker(engine, expire_on_commit=False)
client  = httpx.AsyncClient(timeout=httpx.Timeout(2.0))
log     = structlog.get_logger()

@asynccontextmanager
async def lifespan(app: FastAPI):
    log.info("starting", version=__version__)
    yield                                   # the app serves here
    await client.aclose()                   # reverse order, like Go's defer
    await engine.dispose()

app = FastAPI(title="payments", version=__version__, lifespan=lifespan)
app.include_router(orders.router)

# uvicorn main:app                     — development
# uvicorn main:app --workers 4         — production: uvicorn manages its own workers

# the supply-chain job, in about six lines of CI
- run: uv sync --frozen                # fail if the lock is stale
- run: uv run pip-audit                # known CVEs in the whole tree
- run: uv run ruff check --select S    # shell=True, weak hashes, secrets
- run: uv run mypy .
- run: uv run pytest --cov --cov-fail-under=80

# and in the Dockerfile: resolve in one stage, run in a slim one
FROM python:3.13-slim AS runtime
COPY --from=build /app/.venv /app/.venv
USER 1000                              # never root

# before — every flag from the diagram above, in one eight-line function
def process(data):
    result = []
    for i in range(len(data)):
        if data[i] != None:
            if data[i]["status"] == "ok":
                result.append(data[i]["value"] * 2)
    return result

# after — same behaviour; the name says what it returns and the types hold
def doubled_ok_values(rows: Iterable[Row | None]) -> list[int]:
    return [r.value * 2 for r in rows if r is not None and r.status == "ok"]

# what actually changed:
#   range(len(x))  → iterate the thing        != None → is not None
#   nested ifs     → one filter               dicts   → a typed record
#   "process"      → a name that says the answer, not the verb

# models.py — the contract, written once
from decimal import Decimal
from typing import Annotated, Literal, Self
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator

Money = Annotated[Decimal, Field(gt=0, max_digits=12, decimal_places=2)]
Sku   = Annotated[str, Field(pattern=r"^[A-Z]{3}-\d{4}$")]

class LineItem(BaseModel):
    sku: Sku
    qty: Annotated[int, Field(ge=1, le=100)]
    price: Money

class CreateOrder(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)   # unknown key = error

    email: EmailStr
    currency: Literal["GBP", "EUR", "USD"]
    items: Annotated[list[LineItem], Field(min_length=1, max_length=50)]
    note: str | None = None

# two kinds of rule: one field, or the whole object
    @field_validator("note")
    @classmethod
    def strip_and_bound(cls, v: str | None) -> str | None:
        return v.strip()[:500] if v else None

    @model_validator(mode="after")                # runs once, after coercion
    def within_unattended_limit(self) -> Self:
        if sum(i.qty * i.price for i in self.items) > Decimal("10000"):
            raise ValueError("order exceeds the unattended limit")
        return self

# using it — the only three calls you need day to day
try:
    order = CreateOrder.model_validate(request_json)   # dict  -> object
except ValidationError as e:
    return 422, {"errors": e.errors(include_url=False)}

order.items[0].price * order.items[0].qty     # Decimal arithmetic, exact
payload = order.model_dump(mode="json")       # object -> JSON-safe dict
CreateOrder.model_json_schema()               # the OpenAPI schema, free

# deps.py — one session and one transaction per request
async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:      # closes, returns to the pool
        async with session.begin():            # COMMIT on clean exit,
            yield session                      # ROLLBACK on any exception

# routes.py — HTTP shapes in, HTTP shapes out, nothing else
router = APIRouter(prefix="/orders", tags=["orders"])

class OrderCreated(BaseModel):        # response contract ≠ request contract
    id: UUID
    status: Literal["accepted"]
    total: Decimal

@router.post("", status_code=201, response_model=OrderCreated)
async def create_order(
    body: CreateOrder,                                  # validated already
    idem_key: Annotated[str, Header(alias="Idempotency-Key")],
    session: Annotated[AsyncSession, Depends(get_session)],
) -> OrderCreated:
    try:
        order = await orders.place(session, body, idem_key)   # INSERT inside session.begin_nested()
    except orders.DuplicateKey as e:
        order = await orders.get(session, e.existing_id)  # same answer
    except orders.OutOfStock as e:
        raise HTTPException(409, str(e)) from e           # keeps the cause
    return OrderCreated(id=order.id, status="accepted", total=order.total)

# client.py — one client for the process, not one per call
TIMEOUT = httpx.Timeout(connect=1.0, read=2.0, write=2.0, pool=1.0)
LIMITS  = httpx.Limits(max_connections=100, max_keepalive_connections=20)

client = httpx.AsyncClient(base_url="https://ledger.internal",
                           timeout=TIMEOUT, limits=LIMITS)
# a fresh AsyncClient per call throws away the connection pool and the
# TLS session, and turns a 2 ms call into a 40 ms one

class Retryable(Exception): ...      # 5xx, timeouts: worth another try
class Permanent(Exception): ...      # 4xx: retrying cannot help

@retry(stop=stop_after_attempt(3),
       wait=wait_exponential_jitter(initial=0.1, max=1.0),   # jitter matters
       retry=retry_if_exception_type((httpx.TransportError, Retryable)),
       reraise=True)
async def post_entry(entry: LedgerEntry, idem_key: str) -> LedgerRef:
    r = await client.post("/entries", json=entry.model_dump(mode="json"),
                          headers={"Idempotency-Key": idem_key})
    if r.status_code >= 500:
        raise Retryable(f"ledger {r.status_code}")
    if r.status_code >= 400:
        raise Permanent(r.text[:200])
    return LedgerRef.model_validate(r.json())   # their JSON is untrusted too

# bounded fan-out, partial failure allowed
async def enrich_all(ids: list[str], *, limit: int = 20) -> list[Result | BaseException]:
    sem = asyncio.Semaphore(limit)

    async def one(oid: str) -> Result:
        async with sem:                        # the bound, not the loop
            async with asyncio.timeout(2.0):   # per item, not per batch
                return await enrich(oid)

    return await asyncio.gather(*(one(i) for i in ids), return_exceptions=True)
    # return_exceptions=True: one bad id does not lose the other 199 results

# all-or-nothing instead: 3.11+ TaskGroup cancels siblings on first failure
async def must_all_succeed(ids: list[str]) -> dict[str, Result]:
    async with asyncio.TaskGroup() as tg:
        tasks = {i: tg.create_task(enrich(i)) for i in ids}
    return {i: t.result() for i, t in tasks.items()}

# the exit raises ExceptionGroup — handle by type, not by position:
try:
    results = await must_all_succeed(ids)
except* TimeoutError as eg:
    log.warning("slow upstream", count=len(eg.exceptions))
except* ValidationError as eg:
    log.error("bad payload", count=len(eg.exceptions))

# each stage is lazy; nothing is read until the for loop pulls
def rows(path: str) -> Iterator[dict[str, str]]:
    with gzip.open(path, "rt", newline="") as fh:      # decompresses as it reads
        yield from csv.DictReader(fh)

def valid(raw: Iterable[dict[str, str]]) -> Iterator[Order]:
    for i, r in enumerate(raw, start=2):               # line 1 is the header
        try:
            yield Order.model_validate(r)              # pydantic, at the edge again
        except ValidationError as e:
            log.warning("skipped row", line=i, errors=e.error_count())

# the driver: one transaction per batch, so a failure loses 1000 rows, not 8 GB
from itertools import batched            # 3.12+; recipe in the docs before that

totals: Counter[str] = Counter()
for n, batch in enumerate(batched(valid(rows("orders.csv.gz")), 1000), start=1):
    with session.begin():
        session.execute(insert(OrderRow), [o.model_dump() for o in batch])
    totals.update(o.currency for o in batch)
    if n % 100 == 0:
        log.info("progress", batches=n, rows=n * 1000, totals=dict(totals))

import pytest
from app.orders import Order, parse_qty

@pytest.fixture(scope="module")
def store(tmp_path_factory):
    path = tmp_path_factory.mktemp("data") / "orders.db"
    s = Store(path)          # setup: once per test file
    yield s                  # the tests in this module run here
    s.close()                # teardown: runs even if a test fails

@pytest.mark.parametrize(
    "raw, want",
    [("3", 3), (" 3 ", 3), ("03", 3), ("", None), ("x", None)],
    ids=["plain", "padded", "leading-zero", "empty", "not-a-number"],
)
def test_parse_qty(raw, want):
    assert parse_qty(raw) == want

def test_rejects_negative_quantity(store, monkeypatch):
    monkeypatch.setattr("app.orders.now", lambda: FIXED_TIME)
    with pytest.raises(ValueError, match="quantity"):
        store.add(Order(qty=-1))

# any library module: take a logger, never configure one
log = logging.getLogger(__name__)

def charge(order_id: str) -> None:
    log.info("charging", extra={"order_id": order_id, "req": request_id.get()})

# main.py: configure once, fail fast, drain on SIGTERM
def main() -> None:
    cfg = Config.from_env()          # SystemExit now beats a 500 later
    logging.basicConfig(level=cfg.log_level, handlers=[JsonHandler(sys.stdout)])
    server = make_server(cfg)

    stopping = threading.Event()

    def on_sigterm(signum, frame) -> None:
        ready.clear()                # 1. readiness starts failing
        stopping.set()               # 2. flip a flag -- a handler must never block

    signal.signal(signal.SIGTERM, on_sigterm)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    stopping.wait()                  # main thread parks here until SIGTERM
    time.sleep(5)                    # 3. keep serving while the LB notices
    server.shutdown()                # 4. stop accepting, drain in flight

# the sixty seconds before you open the pull request
uv run ruff format . && uv run ruff check --fix .   # half the list, automatically
uv run mypy .
uv run pytest -q -x --ff              # --ff: last run's failures first
uv run pytest --cov --cov-report=term-missing | tail -20

# then read your own diff — these three greps catch what the tools do not:
git diff --cached | grep -nE 'print\(|breakpoint\(|TODO|XXX'
git diff --cached | grep -nE 'except (Exception|BaseException):\s*$'
git diff --cached | grep -nE 'def \w+\([^)]*=\s*(\[\]|\{\}|set\(\))'

shortlink/
  pyproject.toml          project metadata, dependencies, and every tool's config
  uv.lock                 the resolved, hashed dependency graph. Committed.
  .env.example            every variable the app reads, with safe placeholders
  compose.yaml            Postgres and Redis for local work, nothing else
  Dockerfile              multi-stage; the runtime stage has no build tools
  Makefile                the four commands everyone needs, named the same way

  src/
    shortlink/
      __init__.py
      main.py             the app factory and its lifespan. No business logic.
      config.py           settings, read once from the environment
      domain.py           the rules. Imports nothing from this package.
      schemas.py          the wire contract: what the API accepts and returns
      errors.py           one exception hierarchy, one handler
      db.py               engine, session factory, the session dependency
      models.py           SQLAlchemy tables
      repository.py       the only module that writes SQL
      service.py          use cases. Knows the domain, not HTTP.
      api.py              routers, status codes, dependencies
      security.py         API keys and the rate limiter
      observability.py    logging, tracing, metrics, health

  migrations/             Alembic: env.py plus one file per schema change
    versions/
      0001_create_links.py

  tests/
    conftest.py           fixtures: a real database, an in-process client
    test_domain.py        pure, fast, no I/O
    test_api.py           through the app, against a real Postgres
    test_errors.py        the failure contract, which is also an API

  .github/workflows/ci.yaml

That table is the whole architecture. It is also enforceable: a single ruff rule fails the build when an inner layer reaches outward, which is worth more than any diagram, because diagrams do not fail pull requests.

# pyproject.toml -- one file for the package, the dependencies and every tool.
[project]
name = "shortlink"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "pydantic>=2.9",
    "pydantic-settings>=2.5",
    "sqlalchemy[asyncio]>=2.0.35",
    "asyncpg>=0.30",
    "alembic>=1.13",
    "redis>=5.1",
    "structlog>=24.4",
    "uvicorn[standard]>=0.31",
]

[dependency-groups]                 # PEP 735: dev deps that are NOT shipped
dev = [
    "pytest>=8.3",
    "pytest-asyncio>=0.24",
    "httpx>=0.27",
    "testcontainers[postgres]>=4.8",
    "mypy>=1.11",
    "ruff>=0.6",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 100
src = ["src", "tests"]

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "S", "ASYNC", "TID"]
# S101 is "assert used" -- the point of a test file.
per-file-ignores = { "tests/*" = ["S101"] }

[tool.ruff.lint.flake8-tidy-imports.banned-api]
# The layering rule from the table above, enforced. Inner layers import
# outward over somebody's dead body, and now over a failing build.
"fastapi".msg = "domain/, repository/ and service/ must not import FastAPI"

[tool.mypy]
strict = true
files = ["src", "tests"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
addopts = "-q --strict-markers"

# Makefile -- four verbs, the same four in every repository you own.
# The value is not the shortening; it is that CI and a new joiner run
# identical commands, so "works on my machine" has one less place to hide.
.PHONY: install fmt check test run

install:
	uv sync --all-groups          # exact versions from uv.lock, dev included

fmt:
	uv run ruff format .
	uv run ruff check --fix .

check:                            # what CI runs, runnable locally
	uv run ruff format --check .
	uv run ruff check .
	uv run mypy

test:
	uv run pytest

run:
	uv run uvicorn shortlink.main:app --reload

# src/shortlink/config.py
from functools import lru_cache
from typing import Literal

from pydantic import Field, PostgresDsn, RedisDsn, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",              # local convenience only; absent in production
        env_prefix="SHORTLINK_",      # so a stray DATABASE_URL cannot leak in
        extra="forbid",               # a typo in a variable name fails the boot
    )

    environment: Literal["local", "staging", "production"] = "local"
    # No default. A missing DSN must stop the process, not silently point at
    # localhost -- which is exactly how a staging deploy ends up writing to a
    # database that does not exist and passing its own health check.
    database_url: PostgresDsn
    redis_url: RedisDsn

    base_url: str = "http://localhost:8000"
    db_pool_size: int = Field(default=10, ge=1, le=100)
    rate_limit_per_minute: int = Field(default=60, ge=1)
    log_level: str = "INFO"

    @field_validator("base_url")
    @classmethod
    def _no_trailing_slash(cls, v: str) -> str:
        # Every short link is base_url + "/" + code. One trailing slash here
        # produces a double slash in every URL the service has ever issued.
        return v.rstrip("/")

    @property
    def debug(self) -> bool:
        return self.environment == "local"


@lru_cache
def get_settings() -> Settings:
    """Read the environment exactly once per process.

    Cached because settings are immutable for the life of the process, and
    because FastAPI will call this as a dependency on every request.
    """
    return Settings()  # type: ignore[call-arg]  # values come from the environment

# .env.example -- committed. The real .env is not, and .gitignore says so.
#
# This file is documentation that cannot go stale: `extra = "forbid"` means a
# variable the app does not know about stops the boot, and a missing one
# without a default does too. So the set of names here is the set it reads.

SHORTLINK_ENVIRONMENT=local
SHORTLINK_DATABASE_URL=postgresql+asyncpg://shortlink:shortlink@localhost:5432/shortlink
SHORTLINK_REDIS_URL=redis://localhost:6379/0
SHORTLINK_BASE_URL=http://localhost:8000
SHORTLINK_LOG_LEVEL=DEBUG

# src/shortlink/domain.py -- the standard library, and nothing else.
import secrets
from dataclasses import dataclass
from datetime import UTC, datetime

# Base62 without look-alikes. Removing 0/O and 1/l/I costs four characters of
# alphabet and saves a support ticket every time somebody reads a code aloud.
ALPHABET = "23456789abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ"
CODE_LENGTH = 7
BLOCKED_SCHEMES = frozenset({"javascript", "data", "file", "vbscript"})


class DomainError(Exception):
    """Something the caller did. Never a 500."""


class InvalidDestination(DomainError):
    pass


@dataclass(frozen=True, slots=True)
class Link:
    code: str
    destination: str
    owner_id: str
    created_at: datetime
    expires_at: datetime | None = None

    def is_expired(self, now: datetime | None = None) -> bool:
        if self.expires_at is None:
            return False
        # Always compare aware datetimes in UTC. A naive datetime here is the
        # bug that appears twice a year and is blamed on the database.
        return (now or datetime.now(UTC)) >= self.expires_at


def new_code(length: int = CODE_LENGTH) -> str:
    """A short code with ~41 bits of entropy at the default length.

    secrets, not random: these are guessable identifiers for other people's
    content. The collision is handled by a unique index, not by checking
    first -- see repository.py, where the race is impossible rather than
    unlikely.
    """
    return "".join(secrets.choice(ALPHABET) for _ in range(length))


def validate_destination(raw: str) -> str:
    """Normalise and reject. Returns the URL to store, or raises."""
    url = raw.strip()
    scheme, _, rest = url.partition("://")
    if not rest:
        raise InvalidDestination("destination must be an absolute URL")
    if scheme.lower() in BLOCKED_SCHEMES:
        # An open redirector that will emit javascript: is a stored XSS
        # delivery service with your domain's reputation attached.
        raise InvalidDestination(f"scheme {scheme!r} is not allowed")
    if scheme.lower() not in {"http", "https"}:
        raise InvalidDestination("only http and https destinations are allowed")
    if len(url) > 2048:
        raise InvalidDestination("destination is too long")
    return url

# src/shortlink/schemas.py -- the wire contract. Deliberately not the table.
from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field

from shortlink.domain import Link


class CreateLinkRequest(BaseModel):
    # forbid: a client sending {"destinaton": ...} gets a 422 naming the
    # field, instead of a link quietly pointing nowhere.
    model_config = ConfigDict(extra="forbid")

    destination: str = Field(min_length=1, max_length=2048)
    expires_at: datetime | None = None


class LinkResponse(BaseModel):
    code: str
    short_url: str
    destination: str
    created_at: datetime
    expires_at: datetime | None

    @classmethod
    def of(cls, link: Link, base_url: str) -> "LinkResponse":
        # The mapping lives here, in one place. Returning the ORM object
        # directly is how a password hash ends up in a JSON response the
        # first time somebody adds a column.
        return cls(
            code=link.code,
            short_url=f"{base_url}/{link.code}",
            destination=link.destination,
            created_at=link.created_at,
            expires_at=link.expires_at,
        )

# src/shortlink/models.py -- SQLAlchemy 2.0 typed declarative mapping.
from datetime import datetime

from sqlalchemy import DateTime, Index, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class LinkRow(Base):
    __tablename__ = "links"

    # The short code is the natural key and the only thing ever looked up by,
    # so it is the primary key. A surrogate id here would buy an index nobody
    # queries and a join nobody needs.
    code: Mapped[str] = mapped_column(String(16), primary_key=True)
    destination: Mapped[str] = mapped_column(String(2048), nullable=False)
    owner_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)

    # server_default, not default: the database clock is the one clock every
    # writer agrees on, including a migration or a manual INSERT.
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # Idempotency: the same key from the same owner must return the first
    # link, not create a second. Enforced by the database, because two
    # concurrent requests will pass any check you write in Python.
    idempotency_key: Mapped[str | None] = mapped_column(String(64))

    __table_args__ = (
        Index("uq_links_owner_idem", "owner_id", "idempotency_key", unique=True,
              postgresql_where=idempotency_key.isnot(None)),
    )


class ApiKeyRow(Base):
    __tablename__ = "api_keys"

    id: Mapped[int] = mapped_column(primary_key=True)
    # The sha256 of the key, never the key. Unique and indexed, because every
    # authenticated request looks a caller up by exactly this value.
    key_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    owner_id: Mapped[str] = mapped_column(String(64), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    # Revoke by setting a timestamp rather than deleting the row: you will
    # want to know which key was used after somebody turns one off.
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

# src/shortlink/db.py -- one engine per process, one session per request.
from collections.abc import AsyncIterator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from shortlink.config import get_settings

settings = get_settings()

engine = create_async_engine(
    str(settings.database_url),
    pool_size=settings.db_pool_size,
    max_overflow=0,          # a hard ceiling: better to queue than to exhaust Postgres
    pool_pre_ping=True,      # one cheap round trip beats a stale-connection 500
    pool_recycle=1800,       # under most proxies, connections do not live forever
)

# expire_on_commit=False: without it, touching any attribute after commit
# triggers a fresh SELECT -- inside a response serialiser, on a closed
# session, which is the classic MissingGreenlet at the worst moment.
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    """The transaction boundary, and the only place commit is called.

    One request, one transaction. Anything raised below -- a domain error, a
    bug, a cancelled request -- unwinds through here and rolls back, so a
    handler can never leave half a write behind.
    """
    async with SessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# migrations/versions/0001_create_links.py
"""create links

Revision ID: 0001
"""
import sqlalchemy as sa
from alembic import op

revision = "0001"
down_revision = None


def upgrade() -> None:
    op.create_table(
        "links",
        sa.Column("code", sa.String(16), primary_key=True),
        sa.Column("destination", sa.String(2048), nullable=False),
        sa.Column("owner_id", sa.String(64), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True),
                  server_default=sa.func.now(), nullable=False),
        sa.Column("expires_at", sa.DateTime(timezone=True)),
        sa.Column("idempotency_key", sa.String(64)),
    )
    op.create_index("ix_links_owner_id", "links", ["owner_id"])
    op.create_index(
        "uq_links_owner_idem", "links", ["owner_id", "idempotency_key"],
        unique=True, postgresql_where=sa.text("idempotency_key IS NOT NULL"),
    )


def downgrade() -> None:
    op.drop_table("links")

# Two rules this file obeys, and both are learned the hard way:
#
#   Migrations are forward-only in production. `downgrade` exists so the
#   test suite can round-trip; nobody runs it against real data at 3am.
#
#   Every migration must be safe to run while the OLD code is still serving.
#   Deploys are not atomic. Adding a NOT NULL column with no default takes an
#   exclusive lock and rejects every insert the running version makes -- so
#   it is three deploys: add nullable, backfill, then enforce.

# src/shortlink/repository.py -- the only module that knows SQL exists.
from datetime import datetime

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from shortlink.domain import Link
from shortlink.models import LinkRow


class CodeCollision(Exception):
    """The generated code was already taken. Caller retries with a new one."""


def _to_domain(row: LinkRow) -> Link:
    return Link(
        code=row.code,
        destination=row.destination,
        owner_id=row.owner_id,
        created_at=row.created_at,
        expires_at=row.expires_at,
    )


class LinkRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get(self, code: str) -> Link | None:
        row = await self._session.get(LinkRow, code)
        return _to_domain(row) if row else None

    async def find_by_idempotency_key(self, owner_id: str, key: str) -> Link | None:
        stmt = select(LinkRow).where(
            LinkRow.owner_id == owner_id, LinkRow.idempotency_key == key
        )
        row = (await self._session.execute(stmt)).scalar_one_or_none()
        return _to_domain(row) if row else None

    async def add(
        self,
        code: str,
        destination: str,
        owner_id: str,
        expires_at: datetime | None,
        idempotency_key: str | None,
    ) -> Link:
        """Insert, or raise CodeCollision if that code is taken.

        This is INSERT ... ON CONFLICT DO NOTHING on the idempotency index,
        which makes a repeated request a no-op at the database rather than a
        second link. If nothing came back, either the code collided or the
        request is a retry -- the caller can tell them apart, and only one of
        them is worth retrying.
        """
        stmt = (
            insert(LinkRow)
            .values(
                code=code,
                destination=destination,
                owner_id=owner_id,
                expires_at=expires_at,
                idempotency_key=idempotency_key,
            )
            .on_conflict_do_nothing(index_elements=["owner_id", "idempotency_key"])
            .returning(LinkRow)
        )
        try:
            row = (await self._session.execute(stmt)).scalar_one_or_none()
        except IntegrityError as exc:
            # The other unique constraint: the primary key. Two writers drew
            # the same random code, which is rare and entirely expected.
            raise CodeCollision(code) from exc

        if row is None:
            # ON CONFLICT swallowed it: this owner already used this key.
            existing = await self.find_by_idempotency_key(owner_id, idempotency_key or "")
            if existing is None:  # pragma: no cover -- only under manual deletion
                raise CodeCollision(code)
            return existing
        return _to_domain(row)

# src/shortlink/service.py -- the use cases. No Request, no HTTPException.
from datetime import UTC, datetime

from shortlink import domain
from shortlink.domain import DomainError, Link
from shortlink.repository import CodeCollision, LinkRepository

MAX_CODE_ATTEMPTS = 5


class LinkNotFound(DomainError):
    pass


class LinkExpired(DomainError):
    pass


class LinkService:
    def __init__(self, repo: LinkRepository) -> None:
        self._repo = repo

    async def create(
        self,
        destination: str,
        owner_id: str,
        expires_at: datetime | None = None,
        idempotency_key: str | None = None,
    ) -> Link:
        clean = domain.validate_destination(destination)

        # Bounded retry on collision. Unbounded would turn a misconfigured
        # short code length into an infinite loop holding a transaction open.
        for _ in range(MAX_CODE_ATTEMPTS):
            try:
                return await self._repo.add(
                    code=domain.new_code(),
                    destination=clean,
                    owner_id=owner_id,
                    expires_at=expires_at,
                    idempotency_key=idempotency_key,
                )
            except CodeCollision:
                continue
        raise RuntimeError("could not allocate a unique code")  # a real 500: page someone

    async def resolve(self, code: str) -> Link:
        link = await self._repo.get(code)
        if link is None:
            raise LinkNotFound(code)
        if link.is_expired(datetime.now(UTC)):
            # Distinct from "not found" on purpose: the caller gets 410 Gone,
            # which tells a crawler to stop asking. 404 does not.
            raise LinkExpired(code)
        return link

# src/shortlink/api.py -- routers, dependencies, status codes. Nothing else.
from typing import Annotated

from fastapi import APIRouter, Depends, Header, Response, status
from fastapi.responses import RedirectResponse
from sqlalchemy.ext.asyncio import AsyncSession

from shortlink.config import Settings, get_settings
from shortlink.db import get_session
from shortlink.repository import LinkRepository
from shortlink.schemas import CreateLinkRequest, LinkResponse
from shortlink.security import ApiKey, require_api_key, rate_limit
from shortlink.service import LinkService

router = APIRouter()


def get_service(session: Annotated[AsyncSession, Depends(get_session)]) -> LinkService:
    return LinkService(LinkRepository(session))


ServiceDep = Annotated[LinkService, Depends(get_service)]
SettingsDep = Annotated[Settings, Depends(get_settings)]


@router.post(
    "/links",
    response_model=LinkResponse,
    status_code=status.HTTP_201_CREATED,
    dependencies=[Depends(rate_limit)],
    summary="Create a short link",
)
async def create_link(
    body: CreateLinkRequest,
    service: ServiceDep,
    settings: SettingsDep,
    caller: Annotated[ApiKey, Depends(require_api_key)],
    response: Response,
    idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
) -> LinkResponse:
    link = await service.create(
        destination=body.destination,
        owner_id=caller.owner_id,
        expires_at=body.expires_at,
        idempotency_key=idempotency_key,
    )
    # 201 without Location is a 201 the client has to guess about.
    response.headers["Location"] = f"{settings.base_url}/{link.code}"
    return LinkResponse.of(link, settings.base_url)


@router.get("/{code}", include_in_schema=False)
async def follow(code: str, service: ServiceDep) -> RedirectResponse:
    link = await service.resolve(code)
    # 307, not 302: 307 preserves the method, and a permanent redirect would
    # be cached by every proxy on earth -- making the link impossible to
    # revoke, expire or measure.
    return RedirectResponse(link.destination, status_code=status.HTTP_307_TEMPORARY_REDIRECT)

# src/shortlink/main.py -- the app factory and everything with a lifetime.
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import FastAPI

from shortlink import observability
from shortlink.api import router
from shortlink.config import get_settings
from shortlink.db import engine
from shortlink.errors import install_error_handlers


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Start-up before the first request; shutdown after the last one.

    Everything expensive and long-lived is built here, once. Building a Redis
    client per request is the commonest cause of a service that is fine in
    staging and exhausts file descriptors in production.
    """
    settings = get_settings()
    observability.configure(settings)
    app.state.redis = redis.from_url(str(settings.redis_url), decode_responses=True)
    try:
        yield
    finally:
        # Ordered, and always runs: an unclosed pool keeps the process alive
        # past SIGTERM and turns a rolling deploy into a five-minute wait.
        await app.state.redis.aclose()
        await engine.dispose()


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(
        title="shortlink",
        version="0.1.0",
        lifespan=lifespan,
        docs_url="/docs" if settings.debug else None,   # not in production
        redoc_url=None,
    )
    install_error_handlers(app)
    observability.install_middleware(app)
    app.include_router(observability.health_router)
    app.include_router(router)
    return app


# uvicorn shortlink.main:app -- a module-level app so the reloader can find it.
app = create_app()

# src/shortlink/errors.py -- one shape for every failure that leaves the app.
import structlog
from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from shortlink.domain import DomainError, InvalidDestination
from shortlink.service import LinkExpired, LinkNotFound

log = structlog.get_logger()

# The one place a domain error becomes a status code. Adding an error means
# adding a row here, which a reviewer can see, rather than a raise buried in
# a handler, which they cannot.
STATUS_BY_ERROR: dict[type[DomainError], int] = {
    InvalidDestination: status.HTTP_422_UNPROCESSABLE_ENTITY,
    LinkNotFound: status.HTTP_404_NOT_FOUND,
    LinkExpired: status.HTTP_410_GONE,
}


def problem(
    request: Request, status_code: int, title: str, detail: str, **extra: object
) -> JSONResponse:
    """RFC 9457 problem details. A standard shape beats a bespoke one."""
    body = {
        "type": f"https://errors.shortlink.dev/{title.lower().replace(' ', '-')}",
        "title": title,
        "status": status_code,
        "detail": detail,
        "instance": str(request.url.path),
        # The value a support ticket should quote. It is in the logs too.
        "request_id": request.state.request_id,
        **extra,
    }
    return JSONResponse(
        body, status_code=status_code, media_type="application/problem+json"
    )


def install_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(DomainError)
    async def _domain(request: Request, exc: DomainError) -> JSONResponse:
        code = STATUS_BY_ERROR.get(type(exc), status.HTTP_400_BAD_REQUEST)
        # Expected. Logged at info, and it must never page anyone.
        log.info("domain_error", error=type(exc).__name__, status=code)
        return problem(request, code, type(exc).__name__, str(exc))

    @app.exception_handler(RequestValidationError)
    async def _validation(request: Request, exc: RequestValidationError) -> JSONResponse:
        return problem(
            request,
            status.HTTP_422_UNPROCESSABLE_ENTITY,
            "Validation failed",
            "The request body did not match the schema.",
            errors=[
                {"field": ".".join(str(p) for p in e["loc"][1:]), "message": e["msg"]}
                for e in exc.errors()
            ],
        )

    @app.exception_handler(Exception)
    async def _unhandled(request: Request, exc: Exception) -> JSONResponse:
        # A bug. Log the whole thing on our side, tell the caller nothing --
        # a stack trace in a response body is a free map of your dependencies.
        log.exception("unhandled_error", path=request.url.path)
        return problem(
            request,
            status.HTTP_500_INTERNAL_SERVER_ERROR,
            "Internal error",
            "The request could not be completed. Quote the request id if you contact support.",
        )

# src/shortlink/security.py
import hashlib
import hmac
import time
from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shortlink.config import Settings, get_settings
from shortlink.db import get_session
from shortlink.models import ApiKeyRow

bearer = HTTPBearer(auto_error=False)


@dataclass(frozen=True, slots=True)
class ApiKey:
    owner_id: str


def hash_key(raw: str) -> str:
    """What goes in the database. Never the key itself.

    sha256 and not bcrypt on purpose: an API key is already 256 bits of
    randomness that we issued, so there is no low-entropy password to make
    expensive to guess -- and this runs on every single request.
    """
    return hashlib.sha256(raw.encode()).hexdigest()


async def load_key(session: AsyncSession, digest: str) -> ApiKeyRow | None:
    """Look the caller up by the hash, never by the key.

    Indexed on key_hash, so this is one index probe, and the raw key never
    reaches a query, a log line, or a slow-query report.
    """
    stmt = select(ApiKeyRow).where(
        ApiKeyRow.key_hash == digest, ApiKeyRow.revoked_at.is_(None)
    )
    return (await session.execute(stmt)).scalar_one_or_none()


async def require_api_key(
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
    session: Annotated[AsyncSession, Depends(get_session)],
) -> ApiKey:
    unauthorized = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        "invalid or missing credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    if creds is None:
        raise unauthorized

    digest = hash_key(creds.credentials)
    row = await load_key(session, digest)
    # Belt and braces: the lookup already matched on the hash, but comparing
    # the stored value in constant time means a later change to that query --
    # a prefix match, a LIKE, a cache that returns a near miss -- cannot
    # quietly become an authentication bypass.
    if row is None or not hmac.compare_digest(row.key_hash, digest):
        raise unauthorized

    # One identical failure for "no such key" and "revoked key". Telling them
    # apart is free reconnaissance for whoever is working through a list.
    return ApiKey(owner_id=row.owner_id)


async def rate_limit(
    request: Request, settings: Annotated[Settings, Depends(get_settings)]
) -> None:
    """A fixed-window counter in Redis, shared by every worker.

    An in-process dict is not a rate limit: with four workers behind one load
    balancer the effective limit is four times what you configured, and it
    resets on every deploy.
    """
    key = f"rl:{request.client.host if request.client else 'unknown'}:{int(time.time() // 60)}"
    redis = request.app.state.redis

    # One round trip, not two -- and INCR is atomic, so concurrent requests
    # cannot both read 59 and both decide they are under the limit.
    pipe = redis.pipeline()
    pipe.incr(key)
    pipe.expire(key, 120)          # longer than the window: cheap insurance against clock skew
    used, _ = await pipe.execute()

    limit = settings.rate_limit_per_minute
    if used > limit:
        retry_after = 60 - int(time.time() % 60)
        raise HTTPException(
            status.HTTP_429_TOO_MANY_REQUESTS,
            "rate limit exceeded",
            headers={
                "Retry-After": str(retry_after),
                "RateLimit-Limit": str(limit),
                "RateLimit-Remaining": "0",
            },
        )

# src/shortlink/observability.py
import time
import uuid
from collections.abc import Awaitable, Callable

import structlog
from fastapi import APIRouter, FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
from sqlalchemy import text

from shortlink.config import Settings
from shortlink.db import engine

log = structlog.get_logger()


def configure(settings: Settings) -> None:
    """JSON in production, colour locally. Same events either way."""
    renderer = (
        structlog.dev.ConsoleRenderer()
        if settings.debug
        else structlog.processors.JSONRenderer()
    )
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,   # request_id, without passing it around
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            renderer,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(settings.log_level),
        cache_logger_on_first_use=True,
    )


def install_middleware(app: FastAPI) -> None:
    @app.middleware("http")
    async def request_context(
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        # Trust an inbound id so one id spans every service in the call, but
        # generate one when absent -- a request with no id is unsupportable.
        request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
        request.state.request_id = request_id

        # contextvars, not a parameter: every log line anywhere below this,
        # including inside the repository, carries the id without plumbing.
        structlog.contextvars.bind_contextvars(request_id=request_id)
        started = time.perf_counter()
        try:
            response = await call_next(request)
        finally:
            structlog.contextvars.clear_contextvars()

        elapsed_ms = (time.perf_counter() - started) * 1000
        response.headers["X-Request-Id"] = request_id
        log.info(
            "http_request",
            method=request.method,
            # route template, not the URL: /links/{code} aggregates, and the
            # raw path turns every distinct code into its own metric series.
            path=request.scope.get("route").path if request.scope.get("route") else request.url.path,
            status=response.status_code,
            duration_ms=round(elapsed_ms, 2),
        )
        return response


health_router = APIRouter(tags=["health"])


@health_router.get("/healthz", include_in_schema=False)
async def liveness() -> dict[str, str]:
    """Is the process alive? Nothing else.

    It must not touch the database. A liveness probe that checks a dependency
    restarts every pod in the fleet the moment that dependency has a bad
    minute -- turning a brief database blip into a full outage with no
    capacity left to recover.
    """
    return {"status": "ok"}


@health_router.get("/readyz", include_in_schema=False)
async def readiness() -> Response:
    """Should this process receive traffic? Now the dependencies matter."""
    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
    except Exception:
        log.warning("readiness_failed", check="database")
        return JSONResponse(
            {"status": "degraded", "database": "unreachable"},
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        )
    return JSONResponse({"status": "ready"})

# tests/conftest.py -- a real database, a real app, no network.
from collections.abc import AsyncIterator, Iterator

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from testcontainers.postgres import PostgresContainer

from shortlink.db import get_session
from shortlink.main import create_app
from shortlink.models import Base


@pytest.fixture(scope="session")
def postgres_url() -> Iterator[str]:
    # One container for the whole session: starting it is the slow part, and
    # isolation comes from the per-test transaction below, not from restarts.
    with PostgresContainer("postgres:16-alpine", driver="asyncpg") as pg:
        yield pg.get_connection_url()


@pytest.fixture
async def session(postgres_url: str) -> AsyncIterator[AsyncSession]:
    """Each test runs in a transaction that is rolled back afterwards.

    Faster than recreating the schema per test, and it means tests cannot
    leak state into one another however carelessly they are written.
    """
    engine = create_async_engine(postgres_url)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    connection = await engine.connect()
    transaction = await connection.begin()
    factory = async_sessionmaker(bind=connection, expire_on_commit=False)
    async with factory() as s:
        yield s
    await transaction.rollback()
    await connection.close()
    await engine.dispose()


@pytest.fixture
async def client(session: AsyncSession) -> AsyncIterator[AsyncClient]:
    """The real app, with only the session swapped for the test one.

    ASGITransport means no socket is opened and no port is bound, so the
    suite runs anywhere and cannot flake on a port collision.
    """
    app = create_app()
    app.dependency_overrides[get_session] = lambda: session
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

# tests/test_api.py -- through the real app, against a real database.
import pytest
from httpx import AsyncClient


async def test_create_returns_201_with_location(client: AsyncClient, api_key: str) -> None:
    response = await client.post(
        "/links",
        json={"destination": "https://example.com/a"},
        headers={"Authorization": f"Bearer {api_key}"},
    )
    assert response.status_code == 201
    body = response.json()
    assert body["short_url"].endswith(body["code"])
    # The header a client needs and that nobody remembers to assert on.
    assert response.headers["Location"].endswith(body["code"])


async def test_same_idempotency_key_returns_the_same_link(
    client: AsyncClient, api_key: str
) -> None:
    """The test that a mocked database physically cannot perform.

    Two identical requests, one link. The guarantee lives in a partial
    unique index, so only a real Postgres can prove it holds.
    """
    headers = {"Authorization": f"Bearer {api_key}", "Idempotency-Key": "abc-123"}
    first = await client.post("/links", json={"destination": "https://example.com"}, headers=headers)
    second = await client.post("/links", json={"destination": "https://example.com"}, headers=headers)

    assert first.status_code == 201
    assert first.json()["code"] == second.json()["code"]


@pytest.mark.parametrize(
    "destination",
    ["javascript:alert(1)", "data:text/html,<script>", "ftp://example.com", "not a url"],
)
async def test_dangerous_destinations_are_rejected(
    client: AsyncClient, api_key: str, destination: str
) -> None:
    response = await client.post(
        "/links",
        json={"destination": destination},
        headers={"Authorization": f"Bearer {api_key}"},
    )
    assert response.status_code == 422
    # The failure contract is an API too: assert the shape, not just the code.
    assert response.headers["content-type"].startswith("application/problem+json")
    assert response.json()["status"] == 422


async def test_unknown_code_is_404_and_expired_is_410(client: AsyncClient) -> None:
    assert (await client.get("/nope123")).status_code == 404

# Dockerfile -- two stages: one that can compile, one that ships.
FROM python:3.12-slim AS builder

# uv is copied from its own published image rather than pip-installed, which
# keeps this layer tiny and the version pinned by digest upstream.
COPY --from=ghcr.io/astral-sh/uv:0.4 /uv /bin/uv

ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy
WORKDIR /app

# Dependencies before source. Editing a handler must not re-resolve or
# re-download anything -- this one ordering decision is most of your CI time.
COPY pyproject.toml uv.lock ./
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --frozen --no-install-project --no-dev

COPY src/ src/
RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev


FROM python:3.12-slim AS runtime

# A real user, created in the image. Running as root inside a container is
# one kernel bug away from being root on the node.
RUN useradd --create-home --uid 10001 app
WORKDIR /app

COPY --from=builder --chown=app:app /app/.venv /app/.venv
COPY --from=builder --chown=app:app /app/src /app/src
ENV PATH="/app/.venv/bin:$PATH" PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1

USER app
EXPOSE 8000

# Exec form, so uvicorn is PID 1 and receives SIGTERM directly. In shell form
# the shell is PID 1, it does not forward signals, and every deploy waits for
# the 30-second kill timeout while dropping in-flight requests.
CMD ["uvicorn", "shortlink.main:app", \
     "--host", "0.0.0.0", "--port", "8000", \
     "--workers", "4", "--timeout-graceful-shutdown", "20"]

# .github/workflows/ci.yaml -- the same commands the Makefile runs.
name: ci
on:
  push: { branches: [main] }
  pull_request:

concurrency:                        # a new push cancels the old run
  group: ci-${{ github.ref }}
  cancel-in-progress: true

jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
        with:
          enable-cache: true
          cache-dependency-glob: uv.lock

      # --frozen: fail if uv.lock disagrees with pyproject.toml, rather than
      # silently resolving something nobody reviewed.
      - run: uv sync --frozen --all-groups

      - run: make check          # format, lint, types -- identical locally
      - run: make test           # testcontainers uses the runner's Docker

      - name: Build the image
        run: docker build -t shortlink:${{ github.sha }} .