Language Core (Python 3.12-3.14)

Syntax and semantics of modern CPython: PEP 695, pattern matching, exception groups, f-strings and t-strings, dataclass tooling, free-threading, and removals.

Version Map

| Version | Headline features | Status (2026) | |---|---|---| | 3.12 | PEP 695 generics, PEP 701 f-strings, itertools.batched, override, per-interpreter GIL C-API, distutils/imp removed | Supported, practical floor for new code | | 3.13 | Free-threaded build (PEP 703), experimental JIT (PEP 744), TypeIs/ReadOnly/TypeVar defaults, warnings.deprecated, PEP 594 dead batteries removed | Supported, default choice | | 3.14 | Deferred annotations (PEP 649/749), t-strings (PEP 750), free-threading officially supported (PEP 779), except without parens (PEP 758), compression.zstd, concurrent.interpreters | Supported, verify library wheel coverage |

Version floors are deliberate: do not backport these idioms to 3.10/3.11 projects.

PEP 695 Generics

from collections.abc import Iterable

def first[T](items: Iterable[T], default: T | None = None) -> T | None:
    return next(iter(items), default)

class Box[T]:
    def __init__(self, value: T) -> None:
        self._value = value

    @property
    def value(self) -> T:
        return self._value

type Pair[T] = tuple[T, T]
type Matrix[T] = list[list[T]]

• Scoped type parameters replace module-level TypeVar("T"); covariance is inferred from use.
• Bounds and constraints: def f[T: int | str](x: T), class Repo[T: BaseModel].
• Variadic generics: class Array[*Shape], tuple[*Ts], Unpack[Ts].
• type aliases are lazy and introspectable: Pair.__value__, Pair.__type_params__.
• Runtime access: Box.__type_params__; use typing.get_type_hints or annotationlib for annotations.
• Typing depth covers variance, ParamSpec, and defaults.

Anti-patterns

• Keeping module-level TypeVars that "belong" to one generic function or class.
• type X = Any aliases that erase checker signal; prefer a Protocol.
• Using __value__ at runtime for validation; use pydantic TypeAdapter instead.

Pattern Matching

from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

def classify(msg: dict) -> str:
    match msg:
        case {"type": "ping", "seq": int(seq)} if seq >= 0:
            return f"ping {seq}"
        case {"type": "move", "to": Point(x=0, y=y)}:
            return f"axis {y}"
        case [Point(x=x1), Point(x=x2)] if x1 < x2:
            return "ordered pair"
        case _:
            return "unknown"

• Patterns support literals, captures, | alternatives, mappings, sequences, class patterns, guards.
• Mapping patterns are partial: keys not listed are ignored; **_ captures the rest.
• Sequence patterns bind by position; starred names capture remainders ([first, *rest]).
• Class patterns use __match_args__, which dataclasses generate automatically.
• case str() | bytes() is valid; case int | str without call syntax is not a type pattern.
• Use assert_never on the fallthrough for closed unions instead of a bare _ branch (see typing).

Anti-patterns

• Using match as a switch over unrelated side-effect branches; if/elif is clearer.
• Omitting a total case _ for open input; an unmatched value silently does nothing.
• Capturing with a bare name when you meant a constant: case status: matches anything.

Exception Groups and except*

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(0.01)
    if "bad" in url:
        raise ValueError(f"invalid: {url}")
    return url

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("https://ok"))
            tg.create_task(fetch("https://bad"))
    except* ValueError as eg:
        print("value errors:", [str(e) for e in eg.exceptions])
    except* TimeoutError:
        print("timeouts")

• ExceptionGroup/BaseExceptionGroup (3.11) carry multiple exceptions; except* runs once per matching subgroup and re-raises the remainder.
• TaskGroup raises an ExceptionGroup; unwrap deliberately, never except Exception around it.
• eg.subgroup(ValueError) / eg.split() for programmatic handling.
• except* cannot be mixed with bare except in the same try.
• 3.14 allows except ValueError, TypeError: without parentheses (PEP 758); keep parentheses for 3.12/3.13 compatibility.

Anti-patterns

• except Exception: pass around a TaskGroup, which hides sibling failures.
• Logging eg without traversing eg.exceptions, losing root causes.

f-strings and t-strings

PEP 701 (3.12) makes f-strings full expressions: nested same-type quotes, backslashes, comments, and multi-line expressions are legal.

user = {"name": "ada", "id": 7}
print(f"{user["name"]!r:>10}")          # nested double quotes
print(f"{', '.join(map(str, [1, 2]))}") # calls and comprehensions

3.14 adds template strings (PEP 750) for libraries that must process values safely:

from string.templatelib import Interpolation, Template

def render(t: Template) -> str:
    out: list[str] = []
    for part in t:
        out.append(part if isinstance(part, str) else html_escape(str(part.value)))
    return "".join(out)

t = t"<p>{user['name']}</p>"  # Template, not str

• Use t-strings for SQL/HTML/query builders; the library decides escaping instead of the interpreter.
• Never str(template) and inject it into markup without escaping.
• f-string format spec still applies: f"{value:.2f}", !r, !s, !a.

Deferred Annotations (3.14)

PEP 649/749 evaluate annotations lazily on first __annotations__ access, so forward references work without from __future__ import annotations and runtime introspection is cheaper on import.

class Node:
    def __init__(self, child: Node | None = None) -> None:
        self.child = child

from annotationlib import Format, get_annotations
get_annotations(Node.__init__, format=Format.FORWARDREF)

• from __future__ import annotations remains valid; on 3.14 it implies stringized annotations and changes annotationlib results. Libraries must use annotationlib or typing.get_type_hints, not raw __annotations__.
• FastAPI, pydantic, and dataclasses have adapted; pin minimum versions that understand annotationlib (verify upstream release notes).
• If a library reads __annotations__ directly, test it on 3.14 before upgrading.

Dataclasses vs attrs vs Pydantic

| Tool | Best for | Validation | Slots | Perf notes | |---|---|---|---|---| | dataclasses (stdlib) | Internal records, config objects | None built in (__post_init__) | slots=True (3.10+) | Fast instantiation; no conversion | | attrs | Records needing validators/converters | attrs.validators | Default since @define | Mature, very fast; attrs.resolve_types | | pydantic.dataclasses | Validated dataclass-style models | Full pydantic v2 | slots=True supported | Slower than plain dataclasses; JSON via TypeAdapter | | pydantic.BaseModel | API/IO boundaries, settings | Full pydantic v2 | Internal | Ideal at edges, avoid in hot loops |

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True, kw_only=True)
class RetryPolicy:
    attempts: int = 3
    backoff: float = 0.5
    tags: tuple[str, ...] = field(default_factory=tuple)

• Mutable defaults require field(default_factory=...); slots=True blocks accidental attribute growth and reduces memory.
• frozen=True gives __hash__ only if all fields are hashable; equality semantics change.
• Use @dataclass_transform when writing decorators/frameworks that should look like dataclasses to type checkers (3.11+, or typing_extensions).

Anti-patterns

• @dataclass with a mutable default (items: list = []) — ValueError at class definition.
• Reaching for pydantic models for internal hot-path objects; validation dominates cost.
• Mixing attrs and dataclasses in one layer without a reason.

Free-Threaded Interpreter and JIT

| Build | Flag | State | Guidance | |---|---|---|---| | Default CPython | GIL on | Supported | Default for production | | Free-threaded | python3.13t, --disable-gil | 3.13 experimental, 3.14 officially supported (PEP 779) | Benchmark end-to-end; C extensions must declare thread-safety | | JIT | --enable-experimental-jit | Experimental since 3.13 | Do not enable for production; benchmark before believing gains |

• Detect at runtime: sysconfig.get_config_var("Py_GIL_DISABLED"), sys._is_gil_enabled(), PYTHON_GIL=0/-X gil=0 to disable at start on t builds.
• Single-thread throughput on free-threaded builds is typically lower than GIL builds; parallelism only pays for CPU-bound, multi-threaded workloads.
• C extensions are the blocker: a single unmarked extension can force the GIL or crash. Audit wheels with py-free-threading-style compatibility trackers (verify upstream).
• Per-interpreter GIL (PEP 684) is orthogonal; 3.14 adds concurrent.interpreters and InterpreterPoolExecutor for isolated sub-interpreters.
• Prefer processes today unless profiling on free-threaded shows a clear win.

Removals, Deprecations, and Wall-Clock Items

| Item | Change | Action | |---|---|---| | distutils | Removed 3.12 | Use setuptools/hatchling; migrate setup.py | | imp | Removed 3.12 | Use importlib | | Dead batteries: cgi, telnetlib, crypt, pipes, sndhdr, spwd, uu, xdrlib, mailcap, nntplib, ossaudiodev, sunau, aifc, audioop, chunk, imghdr, msilib, nis | Removed 3.13 (PEP 594) | Equivalent PyPI packages | | lib2to3/2to3 | Removed 3.13 | ruff with UP rules | | datetime.utcnow(), utcfromtimestamp() | Deprecated 3.12 | datetime.now(timezone.utc) | | asyncio.get_event_loop() implicit creation | Deprecated; tightened in 3.14 (verify upstream) | asyncio.run, asyncio.Runner, get_running_loop | | multiprocessing fork default | Deprecated pattern; Linux default moving toward forkserver in 3.14 (verify upstream) | Use spawn/forkserver explicitly | | return/break/continue exiting finally | SyntaxWarning in 3.14 (PEP 765) | Restructure control flow | | tarfile extraction | filter="data" default in 3.14 | Keep explicit filter= on older versions |

Small Stdlib Wins

import itertools, tomllib, graphlib

list(itertools.batched(range(5), 2))            # [(0,1),(2,3),(4,)]
with open("pyproject.toml", "rb") as fh:
    cfg = tomllib.load(fh)
graphlib.TopologicalSorter({"a": {"b"}}).static_order()

• functools.cache for unbounded memoization, lru_cache(maxsize=None) equivalent.
• zoneinfo.ZoneInfo("UTC") instead of pytz; datetime is aware-only in new code.
• hashlib.file_digest, secrets.token_urlsafe, itertools.pairwise.
• 3.14: compression.zstd (PEP 784) for zstd without third-party wheels.
• pathlib.Path.walk avoids os.walk string paths.

Migration Notes

• Replace Optional[X], List[X], Dict[K, V] with X | None, list[X], dict[K, V].
• Replace module TypeVar + Generic[T] with PEP 695 syntax on 3.12+; keep old style only if supporting 3.11 or below (typing).
• Replace datetime.utcnow() with aware UTC everywhere; write a test that fails on naive timestamps.
• Run ruff check --select UP before touching semantics; it automates most 3.12+ idioms.
• On 3.14 upgrades, audit annotation-consuming libraries (DI, ORM, serialization) first.

Anti-Patterns

• Catching broad Exception in library code without re-raising or wrapping with context.
• Mutable default arguments (def f(x, acc=[])) and mutable class attributes.
• eval/exec on any input that crosses a trust boundary.
• Using match with side effects in patterns or guards.
• assert for input validation: stripped under -O.
• Ignoring ResourceWarning/DeprecationWarning in CI; both precede removals.

Checklist

• [ ] requires-python declared and CI covers oldest + newest supported minors.
• [ ] No use of removed stdlib modules or deprecated datetime.utcnow().
• [ ] Generics use PEP 695 where the floor allows; no orphan TypeVars.
• [ ] Exception groups handled via except* or explicit subgroup extraction.
• [ ] Annotation-consuming libraries validated on the target interpreter.
• [ ] Dataclass policy documented (stdlib/attrs/pydantic) and applied consistently.
• [ ] Free-threaded/JIT decisions backed by a benchmark, not enabled by default.

Typing and Runtime Validation

Static typing contracts with pyright/mypy plus runtime validation with pydantic v2: strict setups, Protocols, PEP 695/696 generics, TypedDict/Unpack, overloads, exhaustiveness, and escape-hatch policy.

Checker Setup

# pyproject.toml
[tool.mypy]
python_version = "3.13"
strict = true
warn_unreachable = true
enable_error_code = ["redundant-expr", "truthy-bool", "ignore-without-code"]
plugins = ["pydantic.mypy"]

[tool.pyright]
pythonVersion = "3.13"
typeCheckingMode = "strict"
reportMissingTypeStubs = "warning"
reportUnnecessaryTypeIgnoreComment = true

• strict = true flips on disallow_untyped_defs, disallow_any_generics, warn_return_any, no_implicit_optional, and friends. Adopt per-module with [[tool.mypy.overrides]] instead of weakening the global config.
• pyright's strict is the stricter of the two on narrowing and generics; running both in CI is common for libraries. Run one as the gate and the other as advisory if CI time matters.
• Type-check src/ and tests/ separately: mypy src with strict, mypy tests with relaxed disallow_untyped_defs if fixtures fight the checker.
• --strict failures are cheaper to prevent than to retrofit; if adopting on an existing repo, create a baseline and ratchet (allowlist failing modules, then burn it down).
• Do not type-check with python_version older than requires-python; that silently approves APIs the runtime does not have.

Protocols vs ABCs

from typing import Protocol

class Clock(Protocol):
    def now(self) -> float: ...

def elapsed(clock: Clock, started: float) -> float:
    return clock.now() - started

• Protocols are structural: any object with a compatible now satisfies Clock, no inheritance required. Prefer them for dependencies you cannot edit.
• ABCs are nominal: subclassing is explicit and gives register() and shared implementations. Prefer them when you own the hierarchy and want default behavior.
• Variance: read-only protocols are naturally covariant; a method parameter position makes it contravariant. Let the checker infer and do not fight it with TypeVar.
• @runtime_checkable only supports isinstance checks that verify member presence, not signatures. Use it sparingly; it is not a validation mechanism.
• Keep protocols small (one or two methods). Fat protocols are hard to fake in tests and tend to mirror concrete classes.

Anti-patterns

• isinstance(x, Protocol) as a substitute for pydantic validation.
• Inheriting from a Protocol to "implement" it; structural satisfaction is enough.
• Protocols with __init__ — instantiation shape is not part of the contract.

Generics: PEP 695 and Defaults

from collections.abc import Iterable, Iterator

def chunked[T](items: Iterable[T], size: int) -> Iterator[list[T]]:
    batch: list[T] = []
    for item in items:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []

class Repo[ModelT: BaseModel]:  # bound keeps members typed
    def __init__(self, model: type[ModelT]) -> None:
        self.model = model

    def parse(self, raw: bytes) -> ModelT:
        return self.model.model_validate_json(raw)

class Cache[K, V = str]:  # PEP 696 default (3.13+)
    ...

• Bounds (T: BaseModel) are usually better than constraints (T: (int, str)); constraints collapse the type to the union.
• PEP 696 type defaults let callers omit parameters (class Response[DataT = dict]) while libraries keep full genericity. On 3.12 use typing_extensions.TypeVar(default=...); NoDefault marks "no default".
• Type aliases: type Json = dict[str, "Json"] | list["Json"] | str | int | float | bool | None.
• Variadic tuples: def head[*Ts](*args: *Ts) -> tuple[*Ts]; useful for wrappers.
• Runtime access to type parameters is limited; for runtime generics use pydantic TypeAdapter(Repo[User]) or typing.get_args.

Anti-patterns

• Unbounded T used as T() or T.from_x(); the checker cannot prove constructibility.
• Hand-written variance annotations on mutable containers, which are invariant by design.
• Replacing a plain function with a generic abstract class before there are two use cases.

ParamSpec and Concatenate

from collections.abc import Callable
from functools import wraps
from typing import ParamSpec

P = ParamSpec("P")

def logged[**P, R](fn: Callable[P, R]) -> Callable[P, R]:
    @wraps(fn)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        return fn(*args, **kwargs)
    return wrapper

• PEP 695 syntax spells ParamSpec as **P; Concatenate[Request, **P] models decorators that inject a leading argument (for example auth dependencies).
• Always pair *args: P.args with **kwargs: P.kwargs in the same signature.
• functools.wraps preserves __name__; it does not preserve types, so the callable signature must be expressed in annotations.

TypedDict, Unpack, and ReadOnly

from typing import NotRequired, ReadOnly, TypedDict, Unpack

class Job(TypedDict):
    id: str
    tags: ReadOnly[list[str]]
    retry: NotRequired[int]

class Empty(TypedDict, total=False):
    ...

def run(job: Job, **opts: Unpack[Empty]) -> None:
    print(job["id"])

• total=False makes all keys optional; Required[...] marks mandatory ones inside a partial TypedDict. Prefer explicit NotRequired on a total TypedDict for readability.
• ReadOnly[T] (3.13+, typing_extensions earlier) stops mutation through the typed view.
• Unpack[TypedDict] types **kwargs precisely; use it for option bags instead of **kwargs: Any.
• Closed TypedDicts with closed=True / extra_items= (PEP 728) are available in typing_extensions and rolling into newer CPython — verify upstream before relying on it in libraries.
• TypedDict is for static shapes (JSON payloads, kwargs); it performs no runtime validation. Parse with pydantic if the data is external.

Anti-patterns

• def f(**kwargs: Any) for anything with a known option set.
• Mutating a TypedDict key typed ReadOnly.
• Using TypedDict as an ORM/model class with methods; it is a dict shape.

Overloads

from typing import Literal, overload

@overload
def parse(value: str) -> int: ...
@overload
def parse(value: bytes) -> str: ...
def parse(value: str | bytes) -> int | str:
    if isinstance(value, bytes):
        return value.decode()
    return int(value)

• The implementation signature stays untyped for callers (no @overload), and must accept every overload's inputs.
• Prefer Literal mode flags over parallel functions; overloads shine for return-type dependence on input type.
• Avoid calling an overloaded function from inside its own implementation; the checker sees only overloads there. Delegate to a private _parse_impl.
• Overloads are checked for overlap; unreachable or ambiguous overloads are reported under enable_error_code/reportOverlappingOverload.

Self, Exhaustiveness, LiteralString

from enum import Enum, auto
from typing import LiteralString, Self, assert_never

class Builder:
    def with_name(self, name: str) -> Self:
        self._name = name
        return self

class Mode(Enum):
    READ = auto()
    WRITE = auto()

def mode_flag(mode: Mode) -> str:
    match mode:
        case Mode.READ:
            return "r"
        case Mode.WRITE:
            return "w"
        case _:
            assert_never(mode)

def run_query(sql: LiteralString) -> None: ...

run_query("select 1")          # ok
user_input: str = input()
run_query(user_input)          # checker error: possible injection

• Self (3.11+) types fluent APIs and __enter__ without repeating the class name.
• assert_never turns the default branch into an exhaustiveness proof and a runtime tripwire when a new enum member lands; keep match and if/elif totals.
• LiteralString flows through f-strings and str.join but not through arbitrary str input; use it on SQL/shell/template sinks to make injection a type error.
• TypeIs (3.13+) narrows both branches like isinstance, unlike TypeGuard, which only narrows the positive branch. Use TypeIs for predicates that return bool.

typing_extensions Policy

| Need | Use | |---|---| | Feature in the project's floor version | typing from stdlib | | Supporting older majors | typing_extensions | | Preview features not yet in any stable CPython | typing_extensions behind a comment | | deprecated (PEP 702), Doc, TypeForm, NoDefault, ReadOnly, TypeIs | typing_extensions on old floors, stdlib when available |

• Depend on typing_extensions explicitly; never rely on it arriving transitively.
• Pin a floor that contains the symbol you use and watch upstream removals of pre-release aliases. Type-only imports belong under if TYPE_CHECKING: when they would add a runtime import cycle.

Runtime Validation with pydantic v2

from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator

class User(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)

    id: int = Field(ge=1)
    email: str = Field(min_length=3, pattern=r".+@.+")

    @field_validator("email")
    @classmethod
    def normalize_email(cls, value: str) -> str:
        return value.lower()

user = User.model_validate({"id": 1, "email": "A@B.C"})
users = TypeAdapter(list[User]).validate_python([{"id": 1, "email": "a@b.c"}])
raw = user.model_dump_json()

• TypeAdapter validates arbitrary types (list[User], User | None) without a wrapper model; it is the right tool for function boundaries.
• model_validate(..., from_attributes=True) reads ORM/attribute objects; configure ConfigDict(from_attributes=True) or pass per call.
• field_validator(mode="before"|"after") and model_validator(mode="after") cover field and cross-field rules; return Self from model_validator(mode="after").
• @computed_field, @field_serializer, @model_serializer control output shape; model_dump(exclude_unset=True) is essential for PATCH semantics.
• validate_call enforces annotations on ordinary functions; use it at boundaries, not in hot loops.
• Setting strict=True disables coercion ("1" is not int); default lax mode is usually right for HTTP/JSON, strict for internal FFI.
• ConfigDict(extra="forbid") prevents silent payload drift; log the resulting ValidationError.errors() without echoing secrets.

Anti-patterns

• Pydantic v1 spellings: parse_obj, .dict(), .json(), inner class Config, @validator.
• Models with side effects in validators (DB calls, network) — validation must be pure.
• Any fields that defeat the point of the model; use discriminated unions (Field(discriminator=...)) for polymorphic payloads.
• Using a BaseModel where a frozen dataclass suffices in a hot path.

Escape Hatches and Migration

• cast(T, x) only when a runtime invariant cannot be expressed; keep it at boundaries.
• # type: ignore[code] must carry the code and a short reason; enable warn_unused_ignores/reportUnnecessaryTypeIgnoreComment so stale ignores fail CI.
• Treat Any as an import from untyped libraries: wrap once in a typed adapter function, then keep the rest of the codebase Any-free.
• Migration: typing.List/Dict/Optional to builtins and |; module TypeVar to PEP 695; TypeGuard to TypeIs where both branches narrow; pydantic v1 to v2 via the official migration table; Callable[..., X] to Callable[P, X] when a decorator forwards args.

Checklist

• [ ] Strict checker configuration committed and enforced in CI, with a ratcheting baseline if adopted late.
• [ ] Public API fully annotated; Any occurrences reviewed and localized.
• [ ] Protocols used for external/duck-typed dependencies; ABCs for owned hierarchies.
• [ ] Generics use PEP 695; defaults via PEP 696 where callers benefit.
• [ ] TypedDict + Unpack replace **kwargs: Any; ReadOnly protects shared payloads.
• [ ] Enum/union matches end in assert_never.
• [ ] LiteralString on query sinks; external input validated with pydantic v2 and extra="forbid".
• [ ] Ignores/casts have codes and reasons; unused ignores fail the build.

Async and Concurrency

asyncio, structured concurrency, cancellation and timeout semantics, anyio/trio, sync-async boundaries, queues and backpressure, and thread/process/free-threading guidance.

Choosing a Model

| Workload | Model | Rationale | |---|---|---| | Many sockets, HTTP fan-out, WebSockets | asyncio + TaskGroup | Cheap concurrency, low memory per task | | Blocking SDK/DB driver | asyncio.to_thread / anyio thread pool | Keeps the loop free; bounded pool | | CPU-bound batch, image/ML work | Process pool | Bypasses GIL; pay pickling cost | | Framework-portable libraries | anyio | Same code on asyncio and trio | | Strict cancellation and nurseries | trio | Best-in-class structured concurrency | | Shared-state CPU parallelism | Processes first | Free-threading only with benchmark proof |

Rules: an async def function does nothing until awaited or scheduled; the GIL still serializes CPU-bound threads; and "async" never makes a slow query fast.

Entry Points and Loop Hygiene

import asyncio

async def main() -> None:
    await asyncio.sleep(0)

if __name__ == "__main__":
    asyncio.run(main())

• asyncio.run is the only entry point for scripts; use asyncio.Runner when running several coroutines with shared context/debug settings.
• Inside async code use asyncio.get_running_loop(); get_event_loop() in libraries is deprecated and tightened in newer CPython. Never create loops manually in application code.
• asyncio.run cannot be nested; frameworks already own a loop. Use run_coroutine_threadsafe only when crossing from a real thread.
• Enable debug in development: asyncio.run(main(), debug=True) or PYTHONASYNCIODEBUG=1; -X dev surfaces un-awaited coroutines and slow callbacks.
• uvloop is a drop-in policy on Linux/macOS and typically improves throughput; verify it supports your interpreter (including free-threaded builds) before enabling.

Structured Concurrency

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(0.01)
    return url

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch("a"))
        b = tg.create_task(fetch("b"))
    print(a.result(), b.result())

| Aspect | TaskGroup (3.11+) | asyncio.gather | |---|---|---| | Structure | Scoped; block cannot exit while tasks run | Free-floating futures | | Failure | Cancels siblings, raises ExceptionGroup | Raises first (or returns all with return_exceptions=True) | | Result access | Tasks created inside, read after block | Awaited return list | | Partial failure | Explicit except* / subgroup | Easy to silently ignore |

• Prefer TaskGroup. Reach for gather only for simple all-or-nothing fan-out on 3.10.
• gather(..., return_exceptions=True) returns failures as values; if you use it, inspect every result. Otherwise use exceptions.
• Limit concurrency explicitly with semaphores; a TaskGroup over 10k URLs is an outage.

Cancellation Semantics

• asyncio.CancelledError inherits from BaseException; except Exception will not catch it, but bare except: or except BaseException: will. Never swallow it.
• On cancellation, clean up in finally; re-raise after cleanup.
• await inside a cancelled task's finally may immediately re-cancel if the await itself is cancellable; wrap critical cleanup in asyncio.shield(...) or perform it in a separate task.
• asyncio.shield(coro) protects the inner awaitable but the outer await still raises CancelledError; keep a reference to the shielded task to retrieve its result later.
• Cancellation is cooperative: code that never awaits cannot be cancelled, and a task in finally blocks shutdown.
• TaskGroup cancels remaining tasks when one fails, then waits for them to finish before raising. Never assume the group exits early.

async def flush() -> None:
    await asyncio.sleep(0.01)

async def worker() -> None:
    try:
        await asyncio.sleep(60)
    except asyncio.CancelledError:
        raise
    finally:
        await asyncio.shield(flush())

Timeouts

async def get(url: str) -> bytes:
    async with asyncio.timeout(5):
        await asyncio.sleep(0.1)
        return b"ok"

• asyncio.timeout (3.11+) is the modern API; it cancels the enclosed work and raises TimeoutError. asyncio.timeout_at takes a deadline; handle.reschedule() adjusts an active timeout.
• asyncio.wait_for is legacy: it cancels the task but cannot guarantee the task is done when it returns. Prefer timeout.
• Stack timeouts per layer (connect, read, request) rather than one giant outer timeout; HTTP clients have their own timeout configuration.
• A timeout that must not interrupt a critical section should be placed outside a CancelScope(shield=True) / shield boundary, not inside.

Tasks Must Be Owned

The event loop keeps only weak references to tasks; a dropped create_task can be garbage collected mid-flight.

_background: set[asyncio.Task[None]] = set()

def spawn(coro) -> None:
    task = asyncio.create_task(coro)
    _background.add(task)
    task.add_done_callback(_background.discard)

async def shutdown() -> None:
    for task in list(_background):
        task.cancel()
    await asyncio.gather(*_background, return_exceptions=True)

• Track every background task; on shutdown cancel and await with a grace period.
• Name tasks (create_task(coro, name="...")) so all_tasks() output is readable.
• Eager task factory (asyncio.eager_task_factory, 3.12+) starts coroutines synchronously and can cut latency for cache hits; enable per-event-loop only after benchmarking.
• asyncio.get_event_loop_policy().set_task_factory affects all tasks; keep application-wide changes deliberate.

anyio and trio

import anyio

async def worker(name: str) -> None:
    await anyio.sleep(0.01)

async def main() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker, "a")
        tg.start_soon(worker, "b")

anyio.run(main)

• anyio gives one API over asyncio and trio: create_task_group, CancelScope (with .cancel(), shield=True, deadline), fail_after, move_on_after, to_thread.run_sync, from_thread.run, CapacityLimiter, MemoryObjectStream.
• Use anyio when writing libraries intended to run on both backends or when you want cancel scopes without asyncio's TaskGroup differences.
• trio (open_nursery, CancelScope, trio.to_thread.run_sync) is the reference implementation of structured concurrency; choose it when cancellability and deterministic teardown matter more than ecosystem breadth.
• Test anyio code with the anyio pytest plugin:

import pytest

@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param

@pytest.mark.anyio
async def test_concurrent() -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(anyio.sleep, 0)

Sync and Async Boundaries

| From | To | Use | |---|---|---| | async | blocking function | await asyncio.to_thread(fn, *args) | | async | blocking function, bounded pool | loop.run_in_executor(executor, fn, *args) | | async | subprocess | asyncio.create_subprocess_exec | | sync | coroutine, loop running elsewhere | asyncio.run_coroutine_threadsafe(coro, loop) | | async | CPU-bound | ProcessPoolExecutor or a worker queue |

• asyncio.to_thread propagates contextvars; raw run_in_executor does not. Use to_thread unless you need a dedicated executor.
• Never call asyncio.run, loop.run_until_complete, or time.sleep inside a running loop. nest_asyncio is a hack that hides design errors.
• Synchronous DB drivers (psycopg2, MySQLdb, most ORMs' sync sessions) must be offloaded. Async drivers are preferable where mature.
• Offloading does not create capacity: a thread pool has finite workers, and a thread holding the GIL on CPU work still stalls the process. For CPU, use processes.
• Blocking detection: PYTHONASYNCIODEBUG=1, loop.set_debug(True), loop.slow_callback_duration, plus a runtime blocker (for example blockbuster) in tests to fail on accidental blocking calls.

Queues, Semaphores, Backpressure

import asyncio

queue: asyncio.Queue[str] = asyncio.Queue(maxsize=100)

async def produce(items: list[str]) -> None:
    for item in items:
        await queue.put(item)  # blocks when full: backpressure
    await queue.put("__stop__")

async def consume() -> None:
    while True:
        item = await queue.get()
        try:
            if item == "__stop__":
                return
            await asyncio.sleep(0.001)
        finally:
            queue.task_done()

• Bound every queue. Unbounded queues convert overload into memory growth and OOM.
• put_nowait raises QueueFull; decide whether to drop, shed load, or apply backpressure.
• Cap fan-out with asyncio.Semaphore/BoundedSemaphore; share the limiter with the connection pool size so you do not queue inside the pool.
• For streams, use asyncio.StreamReader/writer or async iterators with explicit window limits; SSE/WebSocket producers must also respect slow-consumer limits.
• asyncio.Event, Lock, Condition are not thread-safe; never share them with threads.
• Graceful shutdown: stop accepting work, drain the queue with a deadline, cancel the rest.

Threads, Processes, Free-Threading

| Tool | Use | Caveats | |---|---|---| | threading | I/O-bound work with blocking APIs | GIL serializes CPU; not cancellable | | ThreadPoolExecutor | Bounded offload from asyncio | Default worker count may oversubscribe | | ProcessPoolExecutor | CPU-bound parallelism | Pickling, startup cost, memory duplication | | multiprocessing.shared_memory | Large arrays across processes | Manual lifetime; no automatic cleanup | | Free-threaded CPython 3.14 | CPU threads without GIL | Extension readiness; single-thread slowdown |

• Start methods: spawn is safest and required on macOS/Windows; fork in a multi-threaded process can deadlock (locks held by threads that do not exist in the child). Newer CPython deprecates fork here and moves the Linux default toward forkserver — verify upstream and set the method explicitly.
• Use ProcessPoolExecutor(max_workers=os.process_cpu_count()) (fall back to os.cpu_count()) and send picklable data; workers cannot share ORM sessions or clients.
• For free-threaded builds, benchmark both single-thread and multi-thread throughput; enable only when your native extensions declare GIL-independence. See language core.
• concurrent.futures tasks are not cancellable once running; cancel only affects queued futures.

Debugging Async Systems

• PYTHONASYNCIODEBUG=1 python -X dev app.py catches un-awaited coroutines and slow callbacks.
• "Task was destroyed but it is pending" means a task reference was lost or shutdown did not await tasks.
• asyncio.all_tasks() and task.get_name()/get_stack() dump what is stuck.
• Hangs are usually: a lock never released, a queue awaiting without a producer, a shielded forever task, or a thread pool exhausted by blocking work.
• Load-test concurrency boundaries, not just throughput: measure p99 under saturation to find pool starvation.

Anti-Patterns

• time.sleep, requests, open(), or a sync DB driver inside async def.
• Fire-and-forget create_task without a reference or shutdown strategy.
• except Exception: pass hiding CancelledError side effects or task failures.
• gather(..., return_exceptions=True) with results ignored.
• Unbounded queues, unbounded fan-out, unbounded thread pools.
• Long CPU work in a coroutine "because it is fast".
• Blocking locks (threading.Lock) acquired inside async code.
• Using asyncio.get_event_loop() to create a loop in library code.

Checklist

• [ ] One documented loop entry point; no nested asyncio.run.
• [ ] Structured concurrency (TaskGroup/anyio) used for fan-out.
• [ ] Every task tracked, named, and awaited on shutdown.
• [ ] Cancellation re-raised; cleanup uses shield where required.
• [ ] Timeouts on all network and external calls, layered per hop.
• [ ] Blocking calls offloaded through to_thread or a process pool.
• [ ] Queues bounded; semaphores limit concurrency to pool capacity.
• [ ] Debug mode or blocker enabled in tests; p99 measured under saturation.

Web Frameworks

FastAPI, Litestar, Starlette, and Django 5.x: lifespan, dependency injection, WebSockets, testing, middleware order, error handling, and rate limiting.

ASGI vs WSGI

| Aspect | WSGI | ASGI | |---|---|---| | Concurrency | One blocking request per worker thread | Coroutines, WebSockets, streaming | | Servers | gunicorn, uwsgi, waitress | uvicorn, granian, hypercorn, daphne | | Middleware | WSGI callables | Async callables (implement both for portability) | | Django | Classic wsgi.py | asgi.py, async views and ORM | | Use when | Legacy stack, mature sync deps | New services, streaming, websockets |

Mixing: ASGI apps can call sync endpoints (framework offloads to a threadpool), but WSGI cannot host async views. Migrate endpoint by endpoint and keep one server pipeline.

FastAPI

Lifespan and app state

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    engine = create_async_engine("postgresql+asyncpg://...", pool_size=10)
    app.state.engine = engine
    app.state.sessionmaker = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

• @app.on_event("startup") is deprecated; use lifespan.
• Lifespan runs once per worker process, not once per cluster; do not run cron/leader work there unless you coordinate (or use a dedicated job runner).
• Store pools, clients, and caches on app.state, never module globals, so tests and multiple app instances stay isolated.

Dependency injection

from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    async with request.app.state.sessionmaker() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

@app.get("/users/{uid}")
async def get_user(uid: int, session: SessionDep) -> dict[str, int]:
    return {"id": uid}

• Dependencies are cached per request by default; Depends(fn, use_cache=False) forces re-execution. Do not abuse this for expensive work.
• yield dependencies run teardown after the response; exceptions raised before yield are handled by the framework, exceptions after yield are not.
• Prefer Annotated[..., Depends(...)] aliases over defaults — they compose and are reusable across signatures.
• Keep dependency chains shallow; deeply nested DI is hard to test and hides I/O.
• Use app.dependency_overrides for tests instead of patching internals.

Sync vs async endpoints

• async def runs on the loop: only call async I/O here.
• plain def runs in Starlette's threadpool: safe for blocking libraries, limited by the pool size (default tied to the anyio thread limiter).
• Never mix: an async def calling a blocking ORM stalls every request in the worker.

Background tasks and queues

from fastapi import BackgroundTasks

@app.post("/send")
async def send(background: BackgroundTasks) -> dict[str, bool]:
    background.add_task(send_email, "user@example.com")
    return {"queued": True}

• BackgroundTasks run after the response in the same process; they are lost on crash and scale with web workers. Use for best-effort work only.
• Durable work belongs in Celery/ARQ/Dramatiq (deployment).
• Do not pass an AsyncSession into a background task: it closes with the request.

WebSockets and streaming

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def echo(sock: WebSocket) -> None:
    await sock.accept()
    try:
        while True:
            await sock.send_text(await sock.receive_text())
    except WebSocketDisconnect:
        return

• Authenticate during the handshake (token in query/subprotocol or first message), not after accepting.
• Implement heartbeats and close codes; idle proxies kill silent sockets.
• For streaming responses use StreamingResponse or SSE with a bounded generator and a slow-client timeout; see async and concurrency.

Testing

import httpx
import pytest

@pytest.fixture
async def client() -> httpx.AsyncClient:
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

async def test_get_user(client: httpx.AsyncClient) -> None:
    resp = await client.get("/users/1")
    assert resp.status_code == 200

• httpx.ASGITransport does not run lifespan. Use asgi-lifespan's LifespanManager, or the Starlette TestClient context manager when lifespan is required.
• Override dependencies and databases per test; never hit production pools from unit tests.
• Assert on schemas and status codes, and keep one contract test per route.

Responses and validation

• Set response_model= (or return an annotated model) to filter output fields.
• response_model_exclude_unset=True is useful for PATCH-like shapes.
• Register a RequestValidationError handler to normalize 422 bodies; never echo the raw payload if it can contain secrets.

Litestar

from litestar import Litestar, get

@get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}

app = Litestar(route_handlers=[health])

• Roughly FastAPI-shaped with batteries: signature-based DI (Provide), DTOs, plugins (SQLAlchemy, OpenAPI), built-in rate limiting, caching, and channels.
• Native support for dataclasses, attrs, msgspec, and pydantic models; msgspec schemas are the fastest path when validation needs are simple.
• Prefer Litestar when you want validation/DI/rate limits without assembling third-party glue; prefer FastAPI when ecosystem depth and hiring pool dominate.
• Test with AsyncTestClient/TestClient from litestar.testing; lifecycle and DI overrides are first-class.

Starlette

• Starlette is the ASGI toolkit FastAPI builds on: Route, WebSocketRoute, Request, Response, BackgroundTask, middleware.
• Write middleware as pure ASGI callables. BaseHTTPMiddleware buffers/behaves differently and can break streaming, background tasks, and contextvars.

class RequestIdMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        scope.setdefault("state", {})["request_id"] = "generated"
        await self.app(scope, receive, send)

• Middleware order: the last registered middleware is the outermost layer; CORS, auth, and request logging must be ordered deliberately (outermost first on the request path).
• Pure ASGI middleware must handle lifespan and websocket scopes or explicitly pass them through.
• Use Starlette directly for small internal services; use FastAPI/Litestar for anything with a public contract.

Django 5.x Async

from asgiref.sync import sync_to_async
from django.http import JsonResponse

async def user_detail(request, pk: int) -> JsonResponse:
    user = await User.objects.aget(pk=pk)          # async ORM
    avatar = await sync_to_async(compute_avatar)(user)  # blocking helper
    return JsonResponse({"name": user.name, "avatar": avatar})

• Async views run on ASGI; calling sync ORM code directly raises SynchronousOnlyOperation. Use aget/acreate/aupdate/adelete, async for over querysets, and sync_to_async for code you cannot convert.
• sync_to_async defaults to thread_sensitive=True, serializing onto one thread; that is safe for transactions but becomes a bottleneck. Use thread_sensitive=False only for independent, thread-safe work.
• Never call sync_to_async inside an open async transaction expecting atomicity across the boundary; transactions and connections are per-thread.
• Async ORM support improves with each 5.x minor (bulk operations, aggregates) — verify the exact APIs for your installed minor upstream.
• Serve Django with uvicorn project.asgi:application (or Daphne). Keep wsgi.py if you still run some sync-only deployments.
• N+1 prevention is still select_related/prefetch_related; async does not change it. See data layer.

Error Handling

| Layer | Pattern | |---|---| | FastAPI/Litestar | Framework exception handlers; HTTPException for expected, handlers for domains | | Validation | 422 with field errors (RFC 9457 Problem Details where clients need machine-readable ops) | | Django | handler404/handler500, DRF exception handler if DRF is present | | Unexpected | Log with request id and trace, return generic 500 without internals |

• Distinguish "client error" (4xx, do not alert) from "server error" (5xx, alert).
• Keep exception handlers thin: map domain exception to response, log once at the boundary.
• Never return stack traces or SQL to clients; never str(exc) an unexpected exception in the response body.
• Timeouts and upstream failures should map to 502/503/504, not 500.

Middleware and Cross-Cutting Concerns

Order (outermost to innermost) that works for most services:

1. Request ID / contextvars propagation
2. Access logging and metrics
3. Trusted host / HTTPS redirect / security headers
4. CORS (only for browser clients; be explicit about origins)
5. Authentication
6. Rate limiting
7. Routing/handlers

• Authentication in middleware is coarse; use dependencies when authorization depends on the route's resources.
• CORS preflight must short-circuit before auth or browsers see auth errors.
• Gzip/Brotli at the edge or with GZipMiddleware; do not double-compress.
• Body size limits belong at the proxy and in the app; unbounded JSON parsing is a DoS vector.

Rate Limiting

• Per-process counters do not work with multiple workers or replicas. Use Valkey/Redis (or an API gateway) for shared state.
• Token bucket or sliding window: INCR + EXPIRE is minimal; sorted sets or Lua scripts give smooth windows.
• Return 429 with Retry-After; make limits keyed per API key/IP/route and document them.
• FastAPI: slowapi or a small custom middleware/DI; Litestar has RateLimitConfig built in. Verify current plugin maintenance before adopting.
• Apply stricter limits to login, password reset, and expensive endpoints than to health checks.

Anti-Patterns

• async def endpoint calling requests, a sync ORM, or time.sleep.
• Long work inside the request path when a queue is available.
• Module-level DB engines/clients shared across test runs.
• Returning ORM models directly and relying on implicit serialization.
• In-memory sessions, caches, or rate limits behind multiple workers.
• BaseHTTPMiddleware in front of streaming endpoints.
• Trusting X-Forwarded-For without a configured trusted proxy list.
• Swallowing exceptions in middleware and returning 200.

Checklist

• [ ] Lifespan manages pools/clients and disposes them on shutdown.
• [ ] Every endpoint is deliberately async def or def based on its I/O model.
• [ ] Dependencies shallow, overridable in tests, using Annotated aliases.
• [ ] Background work is either best-effort and documented, or queued durably.
• [ ] WebSocket auth at handshake, heartbeats, close codes, backpressure.
• [ ] Tests run lifespan when needed and never touch shared external state.
• [ ] Middleware order reviewed; CORS preflight unauthenticated; body limits set.
• [ ] Rate limiting is shared across workers and returns Retry-After.
• [ ] Errors mapped to stable status codes; no internals leaked.

Data Layer

SQLAlchemy 2.0 async sessions and transactions, Alembic, pydantic-settings, Django ORM, pandas 3/Polars, caching, connection pooling, and safe migrations.

SQLAlchemy 2.0 Setup

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

engine = create_async_engine(
    "postgresql+asyncpg://user:pw@host/db",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    pool_recycle=1800,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, autoflush=False)

• 2.0 style only: select(Model).where(...), session.scalars(...). The legacy Query API is removed from new code paths.
• expire_on_commit=False avoids implicit lazy refreshes after commit, which raise in async contexts.
• autoflush=False makes writes explicit; flush/commit at the service boundary.
• Use Mapped[...]/mapped_column declarative models; DeclarativeBase is the base class.
• Sync engine + Session remain valid for scripts, data jobs, and Celery workers; mixing sync and async sessions on the same connection is not possible (MissingGreenlet).

Sessions and Unit of Work

async def create_user(session: AsyncSession, name: str) -> User:
    user = User(name=name)
    session.add(user)
    await session.flush()   # assigns PK, sends INSERT
    return user             # caller commits

• One session per request/task ("unit of work"), never a module global. Sessions are cheap; connections are not.
• Commit once at the outer boundary (async with session.begin():) so multi-repository operations stay atomic.
• flush() to get generated keys; commit() to persist; refresh() only when you need DB-side defaults or triggers.
• After a failed flush the session is in a failed state; roll back and rebuild rather than continuing.
• FastAPI dependency pattern in web frameworks; cancel/shutdown must close sessions.

Transactions and Savepoints

async with SessionLocal() as session:
    async with session.begin():
        session.add(User(name="a"))
        async with session.begin_nested():  # SAVEPOINT
            session.add(User(name="b"))

• session.begin() commits on clean exit, rolls back on exception.
• begin_nested() creates a savepoint; roll back the inner block without losing the outer transaction.
• Avoid manual commit() inside repository methods when the caller owns the transaction; it breaks atomicity and complicates tests.
• SELECT ... FOR UPDATE via .with_for_update() for row locks; keep lock ordering consistent to avoid deadlocks.
• Isolation: default is usually right; do not raise isolation level as a substitute for correct locking.

Loading Strategies and the N+1 Problem

from sqlalchemy import select
from sqlalchemy.orm import selectinload

stmt = select(Order).options(selectinload(Order.items)).where(Order.id == order_id)
order = (await session.execute(stmt)).scalar_one()

| Relationship | Strategy | Notes | |---|---|---| | Many-to-one / one-to-one | joinedload | Single query, FK join | | One-to-many / many-to-many | selectinload | Second query with IN, avoids row multiplication | | Deep chains | selectinload(A.b).selectinload(B.c) | Compose explicitly | | Rarely needed | raiseload("*") / lazy="raise" | Fails fast on accidental lazy load |

• Async lazy loading raises MissingGreenlet; treat that error as a design bug, not an invocation problem.
• Set lazy="raise" (or lazy="raise_on_sql") on relationships by default and opt into eager loading per query — this makes N+1 a startup/test failure.
• selectinload with pagination is safe; joinedload + LIMIT on a collection needs selectinload or a subquery to avoid truncated rows.
• Watch query counts in tests: assert the number of statements per endpoint.

Engine, Pooling, and PgBouncer

• Pool math: max_connections >= replicas * (pool_size + max_overflow). Postgres and pgbouncer have hard limits; exceeding them produces intermittent connection errors under load.
• pool_pre_ping=True detects dead connections after failovers; pool_recycle avoids stale connections behind idle timeouts.
• Behind pgbouncer in transaction mode, prepared statements break connections: disable the driver cache (connect_args={"statement_cache_size": 0} for asyncpg, prepare_threshold=None for psycopg3) or run pgbouncer in session mode.
• For serverless/bursty workers, a small pool or NullPool plus an external pooler is usually safer than a large local pool.
• Always set connect/read timeouts (connect_args or driver-level); an unresponsive DB must not hang the event loop forever.

Alembic

• Initialize with the async template: alembic init -t async migrations.
• Set a deterministic naming convention on the metadata so autogenerate and downgrades can address constraints:

from sqlalchemy import MetaData

NAMING = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}
metadata = MetaData(naming_convention=NAMING)

• Autogenerate does not detect everything: server default changes, type changes, enum alterations, check constraints, and some index renames require manual edits. Review every migration.
• Ship schema and data migrations separately; data backfills belong in explicit scripts with batching and progress logging.
• alembic upgrade head runs at deploy start (or a migration job), never Base.metadata.create_all in production.
• op.batch_alter_table for SQLite; use autocommit_block() for index operations that cannot run inside a transaction.
• In CI, alembic upgrade head then alembic check/--autogenerate to detect drift between models and migrations.

pydantic-settings

from functools import lru_cache

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_nested_delimiter="__", extra="ignore"
    )

    database_url: str
    redis_url: str = "redis://localhost:6379/0"
    api_key: SecretStr
    debug: bool = False

@lru_cache
def get_settings() -> Settings:
    return Settings()

• SecretStr prevents accidental logging of secrets; use get_secret_value() explicitly.
• Validate settings at startup; fail fast with a clear aggregate error.
• Prefer environment variables in production, .env only locally; never commit .env.
• Inject Settings rather than importing a module-level singleton, so tests can override.

Django ORM

• select_related("author") for FK/one-to-one joins; prefetch_related("tags") for many relations. Combine with Prefetch(..., to_attr=...) for shaped caches.
• only()/defer() to avoid wide rows; values()/values_list() for read-only reports.
• Large scans: .iterator(chunk_size=...) (sync) / .aiterator(...) (async) instead of materializing all rows.
• Writes: save(update_fields=[...]), bulk_create/bulk_update (verify minor-version async variants upstream), F() expressions for atomic increments.
• select_for_update() inside transaction.atomic() for row locks; never hold locks across network calls.
• Assert query counts in tests with django_assert_num_queries/assertNumQueries; use django-debug-toolbar or django-silk in development to spot N+1 early.
• Async bridging details in web frameworks.

pandas 3 and Polars

| Aspect | pandas 3.x | Polars | |---|---|---| | Execution | Eager (per-operation) | Lazy by default with streaming | | Memory | Row/column blocks, improved with Arrow strings | Columnar Arrow, typically lower | | Strings | PyArrow-backed by default | Arrow-native | | Multi-core | Single-threaded per op (GIL) | Native multithreading | | Ecosystem | Broadest | Growing, strong Parquet/Arrow | | Choose when | Existing pandas code, libraries | New pipelines, large files, performance |

import polars as pl

result = (
    pl.scan_parquet("events/*.parquet")
    .filter(pl.col("kind") == "click")
    .group_by("user_id")
    .agg(pl.len().alias("clicks"))
    .collect(streaming=True)
)

• pandas 3 shifts defaults: Copy-on-Write semantics, Arrow-backed strings, changed inplace behavior. Upgrade by running the suite with the new defaults and fixing chained assignment warnings; see the official migration guide (verify upstream for the exact minor).
• Prefer vectorized expressions over apply/iterrows; keep Python loops out of hot transforms.
• Convert across libraries explicitly (pl.from_pandas, .to_pandas()); do not mix indexes and lazy semantics implicitly.
• Use polars to read and pre-aggregate before loading into pandas/ML libraries when data is large.

Caching and Ephemeral State

from redis.asyncio import Redis

redis = Redis.from_url("redis://localhost:6379/0", decode_responses=True, socket_timeout=0.5)

async def get_user_name(user_id: int) -> str | None:
    key = f"user:v1:{user_id}:name"
    if (cached := await redis.get(key)) is not None:
        return cached
    name = await load_name(user_id)
    await redis.set(key, name, ex=300)
    return name

• Cache-aside with explicit TTLs; no TTL means unbounded memory and stale Forever.
• Stampede protection: distributed lock, single-flight, or probabilistic early refresh. Add jitter to TTLs so keys do not expire in lockstep.
• Invalidate on write (delete key or version the prefix); version keys when the schema changes so old entries cannot poison new readers.
• Serialize with JSON/msgpack; never unpickle cache contents from shared infrastructure.
• Valkey is the Redis-compatible fork; redis-py works with either. Newer clients (valkey-glide) add native async performance — verify maturity upstream.
• A local in-process LRU (cachetools) is per worker and cannot be invalidated across replicas; use it only for immutable, cheap-to-recompute data.
• Cache failures must degrade to the source of truth, with a timeout and a circuit breaker if the cache is hot.

Migration Safety

• Expand/contract pattern: add nullable column, dual-write, backfill in batches, switch reads, add constraints, drop old column in a later release.
• NOT NULL and type changes require a validation scan; add the constraint separately (NOT VALID then VALIDATE CONSTRAINT on Postgres) to limit lock duration.
• Create indexes concurrently on Postgres (postgresql_concurrently=True inside an autocommit block); never build a large index in a transaction that holds locks.
• Backfill in bounded batches with sleep/throttle; monitor replication lag and deadlocks.
• ALTER TABLE ... RENAME breaks running old code; assume mixed-version deploys for the duration of a rollout.
• Review generated SQL (alembic upgrade --sql, Django sqlmigrate), store migrations in version control, and never edit a migration that has already run in production.

Anti-Patterns

• A global engine/session shared by all requests, or session-per-thread without cleanup.
• Lazy loading in async code (MissingGreenlet) hidden by ad-hoc refresh() calls.
• create_all() in production instead of migrations.
• Committing inside repositories mid-business-operation, breaking atomicity.
• Unbounded iterator() without chunk size; SELECT * on wide tables.
• Cache reads/writes without timeout, TTL, or invalidation plan.
• Row-wise pandas loops and apply where vectorization exists.
• Autogenerated migrations merged without review (they can silently drop data).

Checklist

• [ ] Async engine configured with pool math consistent with DB/pooler limits.
• [ ] One session per request/task; expire_on_commit=False; explicit commit boundary.
• [ ] Eager loading explicit; lazy="raise" catches accidental N+1.
• [ ] Alembic naming convention set; migrations reviewed, reversible, and CI-checked.
• [ ] Settings validated at startup with SecretStr for secrets.
• [ ] Cache TTLs, jitter, invalidation, and timeouts defined.
• [ ] pandas/polars choice documented; upgrade path tested for pandas 3 defaults.
• [ ] Migration plan follows expand/contract with backfill and lock analysis.

Testing

pytest layout and configuration, fixtures, parametrization, async tests, Hypothesis, mocking boundaries, Testcontainers, coverage, snapshots, and CI parallelism.

Layout and Configuration

# pyproject.toml
[tool.pytest.ini_options]
minversion = "8.0"
testpaths = ["tests"]
addopts = "-ra --strict-markers --strict-config"
markers = ["integration: requires external services", "slow: nightly only"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = ["error"]

• --strict-markers catches typo'd markers; --strict-config catches config typos.
• filterwarnings = ["error"] turns deprecations into failures before they become removals.
• src/ layout keeps tests importing the installed package, not the working directory (tooling).
• Recommended layout:

project/
  src/pkg/
  tests/
    conftest.py
    unit/test_core.py
    integration/test_db.py
    fixtures/golden/
  pyproject.toml

• Keep tests/ free of production imports by path hacks; install the package editable in dev (uv sync) so imports match production.

Fixtures

from collections.abc import Iterator
from pathlib import Path

import pytest

@pytest.fixture
def tmp_config(tmp_path: Path) -> Iterator[Path]:
    path = tmp_path / "config.toml"
    path.write_text("[app]\ndebug = false\n")
    yield path

• Fixture scope: function (default, isolated), module, session (expensive clients). Wider scopes leak mutable state; make shared fixtures immutable or reset explicitly.
• conftest hierarchy: root tests/conftest.py for universal fixtures, subdirectory conftest for area-specific ones. Avoid hidden autouse behavior.
• Prefer factories as plain functions over deep fixture graphs; fixtures are for lifecycle, factories for data.
• yield fixtures run teardown even on test failure; keep cleanup exception-safe.
• request.addfinalizer for multiple resources; monkeypatch auto-rolls back.
• Use tmp_path/tmp_path_factory, never the repo root or /tmp hardcoded.

Parametrization

import pytest

@pytest.mark.parametrize(
    ("raw", "expected"),
    [pytest.param("1", 1, id="positive"), pytest.param("-2", -2, id="negative")],
)
def test_parse(raw: str, expected: int) -> None:
    assert int(raw) == expected

• Name cases with ids= for readable CI output; use pytest.param(..., marks=...) to xfail or skip individual cases.
• Indirect parametrization lets fixtures vary by case: @pytest.mark.parametrize("backend", [...], indirect=True).
• Do not build giant cross-product parametrizations that explode runtime; split or sample.
• Reproduce a single case with pytest "tests/test_x.py::test_parse[negative]".

Async Tests

• Choose one plugin: pytest-asyncio or anyio. Mixing both causes loop/fixture confusion.
• pytest-asyncio: asyncio_mode = "auto" marks async tests automatically, or use @pytest.mark.asyncio explicitly. Set asyncio_default_fixture_loop_scope to silence the deprecation and make loop scope explicit.
• Loop scopes: function-scoped loops isolate state but re-create clients per test; session-scoped loops share connection pools and context. Be deliberate — a shared loop with a shared session is a common source of cross-test pollution.
• anyio: @pytest.mark.anyio plus an anyio_backend fixture parametrized over asyncio/trio exercises both backends (async).
• Always add pytest-timeout (or asyncio.timeout inside the test) so a hung await fails the suite instead of hanging CI.
• Assert on cancellation behavior explicitly: start a task, cancel it, await, verify cleanup.

Hypothesis

from hypothesis import given
from hypothesis import strategies as st

@given(st.lists(st.integers()))
def test_reverse_twice(xs: list[int]) -> None:
    assert list(reversed(list(reversed(xs)))) == xs

• Property tests complement example tests: round-trips, idempotence, invariants, ordering.
• Reproduce failures with @example(...) (permanent regression) and --hypothesis-seed=... for a one-off replay.
• derandomize=True in CI trades coverage for reproducibility; keep nightly runs randomized.
• st.assume for preconditions; target to steer search toward interesting values.
• Stateful testing with RuleBasedStateMachine is excellent for caches, retries, and session lifecycles.
• Watch health checks: slow example generation or too-large data usually means the strategy is unbounded — constrain sizes.

Mocking Boundaries

| Boundary | Tool | Notes | |---|---|---| | Environment, attributes | monkeypatch | Auto-rollback, no patching internals | | Collaborator calls | pytest-mock mocker | Spies/assertions; keep to contracts | | Time | time-machine or injected clock | Freezing datetime globally is brittle | | Network | respx (httpx), responses (requests), fake server | Mock at HTTP layer, not the method | | External SDK | Thin adapter + fake adapter | Keep vendor types out of domain code |

• Patch where the name is looked up (app.clients.redis, not redis.Redis).
• Prefer fakes over mocks when behavior matters (in-memory repository, fake queue, fake clock). Mocks assert interactions; fakes assert outcomes.
• Never mock the function under test; never assert on call order unless order is the contract.
• One adapter per third-party boundary makes both production code and tests cleaner.

Testcontainers

import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres() -> PostgresContainer:
    with PostgresContainer("postgres:17-alpine") as pg:
        yield pg

• Pin image tags for reproducibility; never latest.
• Combine with a session-scoped engine, run migrations once, then isolate per test via transaction rollback (fast) or truncated tables (simple).
• with_reuse speeds local iteration but creates hidden state; do not reuse in CI.
• Containers replace docker-compose services for integration tests; keep a small "smoke" set for external dependencies that cannot be containerized.
• Budget startup time: mark integration tests and run them on a separate CI job or schedule.

Coverage

[tool.coverage.run]
branch = true
source = ["src"]
parallel = true

[tool.coverage.report]
show_missing = true
fail_under = 85
exclude_also = ["if TYPE_CHECKING:", "raise NotImplementedError"]

• Branch coverage exposes untested else paths; line coverage alone misleads.
• parallel = true plus coverage combine when using xdist; the coverage plugin in pytest-cov does this automatically.
• Gate CI with fail_under and review missing lines in the PR, but do not chase 100%.
• Exclude only generated code, TYPE_CHECKING blocks, and platform shims — with a reason.
• Coverage of tests themselves is noise; measure src/.

Golden and Snapshot Tests

• syrupy for text/JSON/graph snapshots: pytest --snapshot-update, then review diffs in the PR like code.
• Golden files for large artifacts (report output, rendered HTML, OpenAPI schema). Normalize volatile fields (timestamps, UUIDs, ordering) before comparison.
• Never auto-update snapshots in CI; updates are a reviewable change.
• Keep one snapshot per behavior; huge opaque blobs make regressions hard to review.

CI Parallelism and Flakiness

• pytest-xdist: -n auto --dist loadgroup for test suites; loadscope keeps module-scoped fixtures on one worker.
• Database tests need per-worker isolation: derive a database/schema name from the xdist worker_id fixture, or wrap each test in a rollback transaction.
• pytest-randomly reorders and reseeds; keep it on to expose inter-test dependencies and record the failing seed.
• Shard long suites (for example pytest-split) across CI jobs; keep total wall time below the reviewer attention span.
• Fast PR gate: -m "not integration and not slow". Nightly: full run plus Hypothesis with more examples.
• Flaky test policy: quarantine with an owner and issue; retries (pytest-rerunfailures) are an emergency valve, and every retry should be logged and investigated.
• Common flake sources: time.sleep polling, shared ports, wall-clock assumptions, ordering dependencies, unbounded waits on external services.

Anti-Patterns

• Testing implementation details: private methods, call order, or internal attributes.
• time.sleep instead of polling with a deadline or injecting a clock.
• One giant test that asserts ten behaviors; failures become ambiguous.
• Mocking so much that the test only exercises the mocks.
• Unit tests that hit the real network or production-like databases.
• Sharing mutable module state across tests without reset.
• Skipping slow tests forever; keep a scheduled full run.
• Assertions without context on complex data; use pytest diffs and pytest.raises(match=).

Checklist

• [ ] pytest configured in pyproject.toml with strict markers/config and warnings as errors.
• [ ] Tests use src/ layout import semantics; no path hacks.
• [ ] Fixtures scoped deliberately; no hidden autouse state.
• [ ] Async tests use one plugin; loop scope explicit; timeouts enabled.
• [ ] Property tests cover core invariants; regressions have @example.
• [ ] Mocking happens at boundaries; fakes preferred for behavior.
• [ ] Integration tests use pinned Testcontainers images.
• [ ] Coverage uses branch mode with a reasonable CI gate.
• [ ] Suite runs in parallel with per-worker DB isolation; flaky tests are tracked.

Tooling and Packaging

uv/ruff/hatch/pdm workflows, a complete pyproject, lockfile policy, src layout, versioning, building and publishing to PyPI, monorepos, pre-commit, and CI.

Toolchain Map

| Need | Primary | Alternatives | Notes | |---|---|---|---| | Env + lock + run | uv | pdm, poetry | uv is fastest and handles Python installs too | | Lint + format | ruff | flake8 + black + isort | One tool, one config section | | Build backend | hatchling | setuptools, flit, pdm-backend | Hatchling is simple and VCS-version friendly | | Release versioning | hatch-vcs | setuptools-scm | Single source: git tags | | Task runner | uv scripts / just | make, taskipy | Keep CI and local commands identical | | Publish | uv publish | twine, flit publish | Use PyPI trusted publishing |

Standardize on one stack and document it in the README; mixed pip/poetry/pdm projects are a chronic source of environment drift.

uv Workflows

uv init --lib mypkg          # or --app for services
uv add httpx                 # runtime dependency
uv add --group dev pytest    # PEP 735 dependency group
uv sync --frozen             # CI/prod: exact lockfile
uv run ruff check .          # run in the project env
uv run pytest -q
uv python install 3.13       # managed interpreter
uv lock --upgrade            # refresh resolution (PR-reviewed)
uv export --format requirements-txt --no-dev > requirements.txt
uv tool install ruff          # global CLI tools, isolated

• uv sync --frozen fails if uv.lock disagrees with pyproject.toml; always use it in CI and images.
• uv run guarantees the project environment; no manual source .venv/bin/activate in scripts.
• uv export exists for consumers that still need requirements.txt; the lockfile remains the source of truth.
• Use [tool.uv] default-groups to control what uv sync installs by default.

Complete pyproject.toml

[build-system]
requires = ["hatchling>=1.27", "hatch-vcs>=0.4"]
build-backend = "hatchling.build"

[project]
name = "my-service"
description = "Example service"
readme = "README.md"
license = "MIT"
requires-python = ">=3.12"
dynamic = ["version"]
dependencies = [
    "fastapi>=0.115",
    "pydantic>=2.9",
    "sqlalchemy[asyncio]>=2.0.35",
]

[project.optional-dependencies]
postgres = ["asyncpg>=0.30"]

[dependency-groups]
dev = ["pytest>=8", "pytest-asyncio", "ruff>=0.9", "mypy>=1.13"]

[project.scripts]
my-service = "my_service.__main__:main"

[tool.uv]
default-groups = ["dev"]

[tool.hatch.version]
source = "vcs"

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

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP", "B", "C4", "SIM", "RUF", "S", "ASYNC", "PT", "TID", "LOG", "G"]

[tool.ruff.lint.per-file-ignores]
"tests/**" = ["S101"]

[tool.ruff.format]
quote-style = "double"

• license = "MIT" uses SPDX expressions (PEP 639); older tooling requires the legacy license = { text = "MIT" } form — verify your build backend.
• Build-system requirements are pinned with floors; the build must be reproducible from the sdist alone.
• The wheel target must point at src/; without it hatchling guesses and can ship an empty wheel.
• Keep tool config in pyproject.toml; avoid scattering setup.cfg, .flake8, and mypy.ini.
• If you cannot use VCS versioning (no tags in build context), set version explicitly and keep it in exactly one place.

Lockfile and Dependency Policy

• Applications commit uv.lock; deploys install --frozen. Never resolve at deploy time.
• Libraries commit the lockfile for development/CI but must not pin consumers; test against your declared lower bounds (a periodic uv lock --upgrade plus a minimum-versions job).
• Do not commit uv.lock inside published wheels/sdists; it is not part of the distribution.
• Dependency ranges: floors (>=) for libraries; for services, the lockfile provides the exactness (== not needed in metadata).
• Avoid unpinned VCS/Git dependencies in released packages; they break rebuilds. Pin a commit and plan to replace with a release.
• AUDIT: run a dependency vulnerability scan in CI (security).
• PEP 735 [dependency-groups] is the standard place for dev/lint/test tooling; optional-dependencies is for user-facing extras (pip install "pkg[postgres]").

src Layout

project/
  src/my_service/
    __init__.py
    __main__.py
  tests/
  pyproject.toml
  uv.lock

• src/ prevents tests and scripts from importing the working directory by accident; you always test the installed package.
• Flat layout is acceptable for applications, but src is the default for anything with a public API.
• Mark the package with __init__.py (regular) or explicitly name packages in the build backend (namespace packages).
• Keep a single top-level import name matching the distribution, or document the mapping.

Versioning

• One source of truth: git tags (v1.2.3) via hatch-vcs/setuptools-scm, or one version field in pyproject.toml.
• Read it at runtime with importlib.metadata.version("my-service"); do not duplicate a __version__ string by hand.
• Follow semantic versioning; document deprecation policy (deprecate for at least one minor, remove at a major — see language core).
• Pre-releases (1.0.0rc1) must be installable and tested; do not tag a release candidate that CI has not built.
• Tag signing and protected branches are part of release integrity.

Building, Wheels, and sdists

uv build                     # sdist + wheel into dist/
uv build --wheel             # wheel only
unzip -l dist/my_service-0.1.0-py3-none-any.whl   # inspect contents

• Always ship both sdist and wheel for pure Python; ship wheels for every supported platform when there are binary extensions.
• Pure Python wheels are py3-none-any. Binary wheels must target the ABI; consider the stable ABI (abi3) so one wheel spans Python minor versions.
• Build manylinux/musllinux wheels with cibuildwheel in CI; never publish a Linux wheel built on a developer machine.
• Check the wheel contains py.typed if you ship type information, and that no tests, caches, or secrets are inside.
• python -m build remains a valid fallback; twine check dist/* validates metadata and README rendering.

Publishing to PyPI

• Use PyPI trusted publishing (OIDC from GitHub Actions/other CI) rather than stored API tokens. Tokens, if unavoidable, are project-scoped and short-lived.
• Flow: build in CI on a version tag -> publish to TestPyPI -> install-test -> publish to PyPI.
• uv publish or the official pypa/gh-action-pypi-publish action; neither needs a token with trusted publishing.
• Never overwrite a release: yank instead, and release a patch. PyPI files are immutable.
• Verify post-publish: pip install pkg==x.y.z in a clean container and import it.
• Do not publish from a dirty working tree; the build must come from the tagged commit.

Monorepos

# root pyproject.toml
[tool.uv.workspace]
members = ["packages/*"]

• uv workspaces share one virtual environment and one lockfile; members declare internal deps as { workspace = true }.
• CI installs everything with uv sync --all-packages (or syncs only the affected member with --package).
• Keep per-package pyproject.toml metadata (name, version, deps); the workspace root holds only workspace and shared tool config.
• Independent versioning with VCS tags needs tag prefixes (pkg-a-v1.2.3) and per-package release automation; otherwise version the repo as one train.
• Enforce import boundaries with a linter rule (for example TID/import-linter) so packages do not reach into each other's internals.

pre-commit

# revs below are illustrative; pin the current releases and let autoupdate bump them
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.9.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v6.0.0
    hooks:
      - id: check-merge-conflict
      - id: check-yaml
      - id: end-of-file-fixer

• Pin every rev; pre-commit autoupdate is a reviewable change.
• Keep hooks fast; move heavy checks (tests, type checking) to CI.
• CI remains the source of truth; developers can bypass hooks with --no-verify and CI must still fail.
• pre-commit run --all-files for the first adoption and for CI parity.

CI Example

name: ci
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.12", "3.13", "3.14"]
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v5
        with:
          python-version: ${{ matrix.python-version }}
          enable-cache: true
      - run: uv sync --frozen
      - run: uv run ruff check .
      - run: uv run ruff format --check .
      - run: uv run mypy src
      - run: uv run pytest -q

• Action versions above are illustrative; use current pinned majors.
• Cache uv's download cache, not the virtualenv, for cross-version reliability.
• Separate fast gates (lint, types, unit) from integration and build jobs.
• Add a build job that runs uv build and installs the wheel in a clean environment.
• Publish only on tags, after tests pass, with id-token: write for trusted publishing.

Anti-Patterns

• requirements.txt as the only source of truth with no resolver lock.
• Committing .venv, dist/, or __pycache__.
• setup.py logic and setup.cfg split across files in new projects.
• Version duplicated in pyproject.toml, __init__.py, and docs.
• Unpinned git+https:// dependencies in published packages.
• pip install -e . with a build backend that does not support editable installs.
• Publishing from a laptop or without provenance/attestations.
• Editing a released version in place (yank and bump instead).

Checklist

• [ ] pyproject.toml is the single config source; uv.lock committed.
• [ ] requires-python matches CI matrix and code features.
• [ ] src layout with an explicit wheel packages target; py.typed shipped if typed.
• [ ] Version derived from VCS or one field; runtime reads importlib.metadata.
• [ ] Build produces inspectable sdist + wheel; binary wheels built in CI.
• [ ] Publishing uses trusted publishing and TestPyPI first.
• [ ] pre-commit pinned; CI runs lint, format check, types, tests, build.
• [ ] Workspace/monorepo boundaries enforced and release tagging documented.

Performance

Profiling with cProfile/py-spy/scalene, memory analysis with tracemalloc/objgraph, allocation patterns, native extensions, caching layers, free-threaded performance, and benchmarking.

Method

1. Define the metric: latency percentile, throughput, memory ceiling, or startup time.
2. Reproduce with a realistic workload and a stable environment; record a baseline.
3. Profile before changing code; attribute time and memory to lines/functions.
4. Fix the dominant cost, then re-measure against the baseline.
5. Guard the improvement with a benchmark or regression test; keep it in CI when stable.

Most "slow Python" is one of: blocking I/O, an N+1 query, accidental O(n^2), JSON (de)serialization, or a hot loop that belongs in NumPy/Polars. Only after ruling those out does micro-optimization make sense.

Profiling Tools

| Tool | Type | Overhead | Best for | |---|---|---|---| | cProfile | Deterministic call profiler | High | Call counts, cumulative attribution, offline analysis | | py-spy | Sampling profiler | Near zero | Production processes, hangs, C extensions, flame graphs | | scalene | Sampling, CPU+GPU+memory | Low-medium | Finding memory allocations and CPU hotspots together | | perf | Native sampling | Low | Native/C extension hotspots, kernel time | | memray | Allocation tracker | Medium | Allocation flame graphs, leak hunting | | timeit | Micro benchmark | n/a | Isolated expression/function timing |

cProfile

python -m cProfile -o prof.out -m my_service
python -m pstats prof.out

import cProfile
import pstats

with cProfile.Profile() as pr:
    run_workload()

pstats.Stats(pr).sort_stats("cumulative").print_stats(20)

• Sort by cumulative to find subsystems and tottime to find self-time hotspots.
• Overhead distorts I/O-heavy and threaded code; do not trust absolute times from cProfile.
• --callers/--callees in pstats shows who called the hotspot and from where.

py-spy

py-spy top --pid 1234
py-spy dump --pid 1234
py-spy record -o profile.svg --pid 1234 --duration 30 --threads
py-spy record -o profile.svg -- python -m my_service

• Attach to running processes without code changes; ideal for production incidents.
• dump prints all thread stacks, the fastest way to diagnose a hang or deadlock.
• --native includes C extension frames; --subprocesses follows multiprocessing pools; --idle shows tasks waiting instead of only spinning.
• In containers, ptrace must be permitted (SYS_PTRACE, --cap-add=SYS_PTRACE, or --security-opt seccomp=unconfined in development only).

scalene

scalene --html --outfile prof.html my_service.py
scalene --profile-all --cpu --memory my_service.py

• Separates Python versus native time and tracks memory growth sources, including lines that allocate most.
• --profile-all covers threads and subprocesses; overhead is higher than py-spy.
• Use it when you suspect copy-heavy code (pandas, JSON) driving both CPU and RSS.

Memory Analysis

import tracemalloc

tracemalloc.start(25)
run_workload()
for stat in tracemalloc.take_snapshot().statistics("lineno")[:10]:
    print(stat)

• Start early with PYTHONTRACEMALLOC=25 to capture allocation tracebacks without code changes.
• Compare two snapshots (snapshot2.compare_to(snapshot1, "lineno")) to isolate growth during a specific operation.
• objgraph.show_most_common_types() and show_growth() find type-level growth; gc.get_referrers/get_referents trace cycles.
• gc tuning: raising gc.threshold cuts collection pauses but increases peak RSS; disable cyclic GC only around measured hot paths and always restore it.
• RSS growth with a flat Python heap points at C extensions, allocator fragmentation, or arenas. In containers, MALLOC_ARENA_MAX=2 reduces glibc arena bloat; jemalloc/tcmalloc via LD_PRELOAD often lowers steady-state RSS.
• Use weakref.WeakValueDictionary for caches keyed by identity, and break reference cycles in objects with __del__ or long-lived registries.
• Measure deep size with pympler.asizeof; sys.getsizeof misses referenced objects.

Allocation and Data Patterns

• __slots__ (or @dataclass(slots=True)) removes __dict__ overhead: easily 30-50% less memory for attribute-heavy objects.
• Prefer generators and iterators for streaming transforms; a list comprehension materializes everything.
• Build strings with "".join(parts); repeated += is O(n^2) in the worst case and creates garbage.
• Numeric buffers: array.array, numpy.ndarray, struct, or memoryview instead of lists of Python objects (often 10x+ memory difference).
• Reuse objects in tight loops; avoid creating short-lived temporaries in comprehensions that run millions of times.
• dict/set membership instead of list scans; collections.deque for queue ends; heapq for top-k; bisect for sorted lookups.
• functools.lru_cache/cache for pure functions; note that cache is unbounded.

Vectorization and DataFrames

• NumPy turns per-element Python into C loops: replace explicit loops with array expressions, np.where, broadcasting, and reductions.
• Polars (lazy + streaming) usually beats pandas for large columnar transforms and uses all cores without GIL contention; check data layer.
• Avoid apply, iterrows, and object dtype; they reintroduce Python-level loops.
• Chunk large Parquet/CSV reads; convert once to Arrow and compute in Arrow/Polars before moving to pandas or scikit-learn.
• Beware copies: pandas Copy-on-Write (default in 3.x) avoids some, but astype, merge, and reset_index still allocate. Profile with scalene.

Native Extensions

| Tool | Use | Trade-off | |---|---|---| | NumPy/Polars | Numeric and tabular hot paths | No new language; limited to vectorizable work | | Cython | Typed hot loops, C library interop | .pyx build step; discipline required | | mypyc | Compile existing typed modules | Restricted dynamic features; used by mypy itself | | Rust + PyO3 + maturin | New extension modules, memory safety | Rust toolchain and a release wheel matrix | | ctypes/cffi | Bind existing C libraries | cffi is easier to maintain; ctypes needs no build |

• Move only measured hotspots; cross the native boundary once per batch, not per element.
• Release the GIL inside long native calls (nogil in Cython, Python::allow_threads in PyO3) so threads can run concurrently.
• Keep a pure-Python fallback where possible; build wheels for every supported platform or users will compile from source (see tooling).
• mypyc is attractive when the hot code is already strictly typed; expect build and import overhead and validate behavior with the full test suite.

Caching Layers

• Layers, fastest to slowest: in-process memoization (functools, cachetools), local low-latency store (Valkey/Redis), HTTP/CDN, database materialized views.
• Memoize pure functions only; invalidate or version keys for anything mutable.
• Unbounded functools.cache on high-cardinality inputs is a memory leak; use lru_cache with a size or an eviction-aware cache.
• Measure hit rate: a cache below roughly 50% hit rate frequently costs more than it saves in memory and invalidation complexity.
• Precompute at startup for immutable configuration; do not recompute per request.
• Remember per-process caches multiply by worker count and are not invalidated across replicas.

Free-Threaded and JIT Performance

• Free-threaded builds (3.13 experimental, 3.14 officially supported) trade single-thread throughput for multi-core scaling; per-object locking and biased-reference removal cost single-thread speed. Benchmark the actual workload.
• Gains require CPU-bound, thread-parallel work and GIL-independent native extensions; I/O code sees little difference.
• The experimental JIT (3.13+) shows modest, workload-dependent gains; treat as research, not a production lever. Always verify upstream status and measure.
• When comparing builds, include startup time, memory, and tail latency, not just peak throughput. See language core.

Benchmarking

def test_parse(benchmark) -> None:
    assert benchmark(parse, "1") == 1

• pytest-benchmark for micro benchmarks: run with --benchmark-autosave, compare across commits with --benchmark-compare, fail on regressions with a threshold.
• asv for long-term tracking across commits and machines with regression detection and HTML reports; useful for libraries.
• Hygiene: warm up, run multiple rounds, report median/percentiles (not one number), pin CPU frequency where possible, isolate from other work.
• CI runners are noisy; treat differences under roughly 10-20% as noise unless using dedicated hardware. Never block a PR on a micro benchmark that has not demonstrated stability.
• Benchmark the same input shapes/sizes as production; micro benchmarks on tiny inputs rarely predict real behavior.

Anti-Patterns

• Optimizing by intuition or rewriting in C/Rust before profiling.
• Timing single runs with time.time() and calling it a benchmark.
• Benchmarking with assertions disabled or debug builds enabled inconsistently.
• Caching everything, with no eviction or invalidation, then debugging memory growth.
• Ignoring I/O and database round-trips while micro-optimizing pure Python.
• Using eval/exec as a performance trick.
• Assuming async is faster for CPU work; it is not.
• Keeping a hot data structure as a list when a set/dict would make it O(1).

Checklist

• [ ] Performance target defined (p50/p99 latency, throughput, RSS, startup).
• [ ] Baseline recorded with a reproducible workload.
• [ ] Profile captured with py-spy/scalene/cProfile and the hotspot identified.
• [ ] I/O, queries, and serialization ruled out before micro-optimization.
• [ ] Memory measured with tracemalloc/objgraph; leaks and fragmentation addressed.
• [ ] Native extensions only for measured hotspots, with coarse boundaries.
• [ ] Cache hit rate and eviction strategy known.
• [ ] Benchmark committed and stable; regressions gated or tracked.

Security and Observability

Python-specific vulnerabilities (deserialization, injection, secrets), password hashing, supply chain, structured logging, OpenTelemetry, metrics, health checks, and error tracking.

Threat Map

| Vulnerability | Python surface | Primary mitigation | |---|---|---| | RCE via deserialization | pickle, marshal, yaml.load, dill, joblib.load, torch.load | Never deserialize untrusted data; signed/typed formats | | SQL injection | f-string/format SQL, text() misuse, raw cursors | Bound parameters; ORM expressions; allowlist identifiers | | Command injection | os.system, subprocess(shell=True) | Argument lists; shlex.quote when a shell is unavoidable | | Path traversal | os.path.join with user input, archive extraction | Path.resolve() + is_relative_to; extraction filters | | Secrets exposure | .env commits, logs, error reports, image layers | Secret managers, redaction, scanning | | Weak password storage | MD5/SHA1, home-grown schemes, stale passlib | Argon2id or bcrypt with current libraries | | SSRF | requests/httpx to user-controlled URLs | Allowlists, block private/metadata ranges, timeouts | | XML attacks | xml.etree entity expansion | defusedxml; disable external entities | | Timing attacks | == on tokens/HMACs | hmac.compare_digest | | Supply chain | Typosquatting, dependency confusion, stolen tokens | Lockfiles, audits, trusted publishing, SBOM |

Deserialization

• pickle executes arbitrary code by design (__reduce__). No sandbox makes untrusted pickle safe. The same applies to dill, joblib.load, and most torch.load payloads.
• yaml.load without a safe loader is code execution. Use yaml.safe_load; even then, alias expansion can be a DoS — cap document size and disable aliases where possible.
• marshal is for .pyc files, not data exchange; it is not a security boundary.
• torch.load: use weights_only=True where supported and prefer safetensors for model weights (verify upstream support in your framework version).
• Safe interchange: JSON, msgpack, protobuf, CBOR; add HMAC/Ed25519 signatures when integrity matters, and validate the schema with pydantic.
• Cache/session stores: store JSON, not pickled objects, so a compromised store cannot achieve RCE.

Anti-patterns

• pickle.loads(redis.get(key)) "because it is faster".
• yaml.load(stream) without Loader= on config uploaded by users.
• Deserializing model checkpoints from public hubs without provenance and weights-only loading.

Injection

SQL

from sqlalchemy import text

await session.execute(
    text("select id from users where email = :email"), {"email": email}
)

• ORM expression builders parameterize automatically; text() is safe only with bound parameters.
• Identifiers (table/column names) cannot be bound: choose them from an allowlist, never from request input.
• Stored procedures and raw cursor.execute calls need the same discipline; review every string that reaches a database.

OS commands

import shutil, subprocess

path = shutil.which("git")
if path is None:
    raise RuntimeError("git not found")
subprocess.run([path, "clone", "--depth", "1", url], check=True)

• Never shell=True with interpolated input; if a shell is truly required, shlex.quote every value and prefer fixed command templates.
• Validate and canonicalize file paths before passing them to subprocesses.

Templates and HTML

• Jinja2: keep autoescape=True for HTML (framework defaults do this); never render user-controlled template source (Template(user_input) is RCE).
• SandboxedEnvironment reduces, but does not eliminate, risk from untrusted authors.
• Avoid Markup/|safe on user data; sanitize rich text with a dedicated library.

Paths and archives

from pathlib import Path

base = Path("/srv/data").resolve()
candidate = (base / user_path).resolve()
if not candidate.is_relative_to(base):
    raise ValueError("path traversal")

• os.path.join with an absolute component discards the base; resolve and verify containment instead.
• tarfile: pass filter="data" (3.12+) — the default in newer CPython — and still bound extraction size and member count.
• zipfile: validate resolved member paths and refuse absolute or .. entries; cap decompressed size (zip bombs).
• XML: defusedxml for untrusted documents; disable DTD/entity resolution.
• SSRF: allowlist schemes/hosts, resolve DNS and block private ranges plus the cloud metadata address (169.254.169.254), disable redirects or re-validate each hop, set short timeouts.

Anti-patterns

• eval, exec, or ast.literal_eval misused as a parser (use JSON/pydantic).
• Fetching webhook_url supplied by a user without SSRF controls.
• Building shell strings by concatenation, however "internal" the endpoint.

Secrets Management

• Source of truth: a secret manager (Vault, AWS/GCP/Azure secret manager) or the platform's environment injection. .env is a local convenience, never a production secret store.
• Never commit .env; add it to .gitignore, .dockerignore, and secret-scanning hooks (gitleaks/trufflehog) in CI.
• Wrap secrets in SecretStr or a similar type so accidental repr/logging does not leak them; call get_secret_value() explicitly.
• Do not log headers, cookies, tokens, or request bodies containing credentials. Configure redaction in the logger and in error tracking.
• Prefer short-lived credentials: OIDC workload identity, instance roles, dynamic database credentials. Rotate on exposure and on schedule.
• Generate tokens and IDs with secrets, never random.

Password Hashing

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

hasher = PasswordHasher()
stored = hasher.hash(password)
try:
    hasher.verify(stored, password)
except VerifyMismatchError:
    raise

• Argon2id (argon2-cffi) is the default recommendation; tune time/memory to roughly 50-100 ms on production hardware and store the parameters with the hash.
• bcrypt is acceptable with cost >= 12; remember it truncates beyond 72 bytes and needs explicit handling (pre-hash with SHA-256 or reject long inputs).
• hashlib.scrypt/pbkdf2_hmac are acceptable stdlib fallbacks; PBKDF2 iteration counts must track current guidance.
• passlib is unmaintained and incompatible with modern bcrypt releases — do not start new projects on it. Use argon2-cffi directly or a maintained wrapper such as pwdlib (verify upstream).
• Never store passwords reversibly; never log or email them; enforce a strong minimum and rate-limit authentication attempts.
• Compare hashes and tokens with hmac.compare_digest, not ==.

Tokens and Sessions

• JWT: use PyJWT and an explicit algorithms=["RS256"] (or similar) allowlist; reject alg: none; validate exp, nbf, aud, iss; handle clock skew and JWKS rotation.
• python-jose is effectively unmaintained; prefer PyJWT unless you need its specific algorithms.
• Keep JWT payloads minimal; they are signed, not secret. Use opaque server-side sessions when revocation matters.
• Session IDs: secrets.token_urlsafe(32); rotate on login and privilege change; set HttpOnly, Secure, SameSite=Lax/Strict; scope cookies tightly.
• Short access-token lifetimes plus refresh rotation; revoke refresh tokens on reuse detection.

Crypto Hygiene

• cryptography (AES-GCM, Ed25519, X25519) for real cryptography; never implement primitives.
• hashlib.sha256/BLAKE2 for integrity, not passwords; hmac for MACs with compare_digest.
• Keep TLS verification enabled everywhere (httpx, requests, DB drivers); never verify=False outside a documented test.
• Pin and update crypto dependencies; CVEs in these libraries are high impact.

Supply Chain

• Install from a lockfile; use hash checking where the workflow supports it.
• Run pip-audit (or osv-scanner) on the lockfile in CI and fail on known vulnerabilities, with a documented, time-boxed exception process.
• Static analysis: bandit and/or ruff security rules (S) in the lint gate; add semgrep rules for organization-specific policies.
• Generate an SBOM (CycloneDX via cyclonedx-py, or syft) per release and store it with the artifact.
• Use PyPI trusted publishing with build provenance; do not keep long-lived tokens in CI.
• Dependency confusion: prefer a single curated index/proxy (devpi, Artifactory) and avoid mixing --extra-index-url with public indexes on private package names.
• Review new transitive dependencies; typosquatted names are a common entry point.

Structured Logging

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger()
log.info("user_created", user_id=7)

• structlog gives processors and contextvars binding; loguru is simpler with less structured control; stdlib logging with dictConfig remains valid.
• Production: JSON to stdout; development: human-readable console renderer.
• Log events with structured fields, not interpolated prose; keep a stable event name for search/alerting.
• Bind request_id/trace_id with contextvars at the boundary; asyncio.to_thread propagates context, raw executor calls do not.
• Redact secrets/PII centrally (processor/serializer), not per call site.
• Never log full request bodies, auth headers, tokens, or credentials.

OpenTelemetry, Metrics, Health

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("charge_card") as span:
    span.set_attribute("payment.provider", "stripe")

• Auto-instrumentation (opentelemetry-instrument, opentelemetry-instrumentation-fastapi, -sqlalchemy, -httpx) covers most stacks; add manual spans only around domain operations.
• Propagate trace context across queues and RPC: inject/extract with the W3C trace-context propagator; carry trace_id in log fields.
• Sampling: parent-based plus a rate limit in production; never sample errors away. Capture exception events on spans for error correlation.
• Metrics with prometheus-client: counters, histograms, gauges; follow RED (rate/errors/duration) for services and USE for resources.
• Label cardinality is the classic outage: never label with user IDs, raw URLs, or unbounded values. Aggregate into buckets.
• Health endpoints: /healthz (liveness, no dependencies) and /readyz (dependency checks with tiny timeouts and cached results). Probes must not hammer the database or perform migrations.

Error Tracking

• Sentry (sentry-sdk[fastapi]) or equivalent: set environment, release (git SHA), traces_sample_rate, and PII scrubbing (send_default_pii=False plus explicit before_send redaction).
• Capture once at the boundary; avoid duplicate events from both middleware and handlers.
• Add context (request id, route, version) but never credentials or full payloads.
• Alert on new issue types, regressions, and error-rate SLO burn, not on every event.
• Test that scrubbing actually removes your secret fields before enabling in production.

Anti-Patterns

• Unpickling anything that crossed a network or trust boundary.
• subprocess with shell=True and interpolated input.
• SQL built with f-strings; identifiers taken from requests.
• Logging tokens, cookies, passwords, or full request bodies.
• MD5/SHA1 password hashes, or == comparisons of secrets.
• Disabling TLS verification to "make it work".
• Secrets baked into images or committed .env files.
• Unbounded metric labels, log lines without request correlation.
• Extracting archives or fetching URLs supplied by users without validation.

Checklist

• [ ] No unsafe deserialization on untrusted input; signed formats where integrity matters.
• [ ] All SQL parameterized; identifiers allowlisted; no shell interpolation.
• [ ] Path containment and archive extraction validated; SSRF controls on outbound fetches.
• [ ] Secrets from a manager, wrapped, redacted, scanned in CI.
• [ ] Password hashing with a maintained Argon2id/bcrypt implementation and sane parameters.
• [ ] JWT/session handling validates algorithms, lifetimes, audiences; tokens compared in constant time.
• [ ] Lockfile + pip-audit + static security lint + SBOM in the release process.
• [ ] Structured JSON logs with request/trace IDs and centralized redaction.
• [ ] Traces/metrics exported with bounded label cardinality; errors captured without PII.
• [ ] Liveness/readiness separated and cheap; alerting tied to SLOs.

Deployment and Runtime

Container images, uvicorn/granian/gunicorn process models, graceful shutdown, health and readiness probes, resource limits, serverless, background job runners, and the deployment checklist.

Runtime Matrix

| Target | Use when | Watch out for | |---|---|---| | Containers on k8s/ECS | Default for services with long-lived connections and workers | Image size, shutdown, probes, limits | | Serverless (Lambda, Cloud Run) | Bursty/HTTP workloads, scale-to-zero | Cold starts, connection limits, execution time caps | | VM / systemd | Simple internal services, legacy constraints | Manual scaling, no orchestration rollback | | Batch/CLI | Cron, data pipelines | Idempotency, locking, retries |

Pick one primary path and make it boring; every additional runtime multiplies release surface.

Container Images

| Base | Pros | Cons | |---|---|---| | python:3.x-slim (Debian) | glibc, wheel compatibility, debuggable | More packages/CVEs than distroless | | Distroless (gcr.io/distroless/python3) | Minimal attack surface, no shell | Harder to debug, fewer tools | | Alpine | Small | musl: many binary wheels unavailable, compilation, slower allocator | | Hardened minimal (Chainguard/Wolfi) | Small + patched | Vendor dependency |

Use slim for most services; distroless for hardened production. Avoid Alpine wherever NumPy/pandas/ML wheels matter (performance).

FROM python:3.13-slim AS base
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=1
WORKDIR /app

FROM base AS build
# pin to the current stable uv image tag or digest
COPY --from=ghcr.io/astral-sh/uv:0.9 /uv /usr/local/bin/uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project
COPY . .
RUN uv sync --frozen --no-dev

FROM base AS runtime
RUN useradd --create-home --uid 10001 app
WORKDIR /app
COPY --from=build --chown=app:app /app/.venv /app/.venv
COPY --from=build --chown=app:app /app/src /app/src
ENV PATH="/app/.venv/bin:$PATH" PORT=8000
USER app
EXPOSE 8000
CMD ["uvicorn", "my_service.asgi:app", "--host", "0.0.0.0", "--port", "8000"]

• Copy the lockfile before the code so dependency layers cache; --no-install-project keeps code changes from re-resolving dependencies.
• Pin the uv image tag (or digest); verify the current stable tag upstream.
• No compilers or dev headers in the runtime stage.
• .dockerignore: .git, .venv, tests, __pycache__, dist, .env, local data.
• Run as a non-root UID, drop capabilities, and consider a read-only root filesystem.
• Never bake secrets into layers; inject at runtime.

Runtime environment

| Variable | Purpose | |---|---| | PYTHONUNBUFFERED=1 | Immediate logs for platforms that capture stdout | | PYTHONDONTWRITEBYTECODE=1 | No __pycache__ writes in read-only images | | PYTHONFAULTHANDLER=1 | Native crash tracebacks | | MALLOC_ARENA_MAX=2 | Limits glibc arena growth in containers | | TZ=UTC | Consistent timestamps | | PYTHONHASHSEED | Leave random; do not pin unless required |

Log to stdout/stderr, never to files inside the container.

Process Model

| Server | Model | Notes | |---|---|---| | uvicorn (one process) | asyncio | Let the orchestrator scale replicas | | uvicorn --workers N | Pre-fork multiprocess | Simple; limited zero-downtime reload; verify current behavior upstream | | gunicorn + UvicornWorker | Mature pre-fork arbiter | Graceful reload, signals, battle-tested | | granian | Rust HTTP server | Multi-worker, HTTP/2; verify maturity for your stack | | hypercorn | asyncio/trio | HTTP/2 and trio backend |

• Worker count: start from CPU cores, then measure. I/O-bound async services often need fewer workers because each multiplexes; more workers help blocking-library workloads only if threads are exhausted.
• Prefer more replicas over many in-container workers: memory and CPU limits, restarts, and graceful shutdown are per-container.
• --preload (gunicorn) can reduce memory with copy-on-write, but defers per-worker initialization; test startup time and failure modes.
• Do not run migrations, cache warming, or cron in the web process; use init jobs or dedicated workers.
• Use uvloop (asyncio) or the server's native loop where supported and verified.

Graceful Shutdown

• On SIGTERM: stop accepting connections, finish in-flight requests, close pools, cancel background tasks, then exit.
• Bound the wait: uvicorn --timeout-graceful-shutdown, gunicorn --graceful-timeout, granian equivalents. Requests exceeding the deadline are terminated.
• Implement shutdown in lifespan: cancel and await tracked tasks, dispose the engine, flush buffers (async).
• Align layers: load balancer deregistration delay + k8s terminationGracePeriodSeconds + app graceful timeout, in that order, with margin.
• Queue workers: process the current message to completion within the limit, otherwise requeue (with acks_late and idempotency).
• Long-lived WebSockets need an explicit close/reconnect path during deploys.

Health and Readiness

| Endpoint | Checks | Used by | |---|---|---| | /healthz (liveness) | Process responsive, event loop healthy | k8s liveness/startup probes | | /readyz (readiness) | DB ping, cache ping, migrations applied | Load balancer, k8s readiness | | /version | Build SHA, version | Humans, deploy verification |

• Liveness must not check dependencies: a slow database would restart every replica.
• Readiness checks should have tight timeouts (tens of milliseconds to low seconds) and cache results briefly so probes do not hammer dependencies.
• startupProbe gives slow imports/migrations time without loosening liveness.
• Degrade gracefully: if the cache is down but the DB can serve, stay ready and alarm.
• Never expose internal hostnames, credentials, or stack traces from these endpoints.

Resource Limits and Scaling

• Set requests for steady state and memory limits with headroom for GC and large requests; exceeding the memory limit means OOMKill, not a graceful error.
• CPU limits throttle workers mid-request; Python needs CPU for GC, JSON, and TLS. Avoid aggressive CPU limits on latency-sensitive services.
• Connection pools are per worker: total DB connections = replicas x workers x pool size. Size pools down as you scale out (data layer).
• Autoscale on the signal that saturates first: RPS/queue depth for I/O-bound, CPU for CPU-bound; use p99 latency and saturation to validate.
• File descriptor and connection timeouts prevent silent leaks under load.

Serverless

AWS Lambda

• Use Mangum to adapt an ASGI app, or a minimal handler for small functions. Container images up to the platform limit, otherwise zip with a slim dependency set.
• Cold starts: minimize module-level imports, lazy-load heavy libraries, keep the package small, and use provisioned concurrency where latency SLOs demand it (SnapStart support for Python varies — verify upstream).
• Database connections: use RDS Proxy/connection pooler; serverless concurrency will otherwise exhaust connections.
• Execution ceiling and /tmp ephemerality: externalize state, persist results, make retries idempotent.
• Add Powertools for structured logging, tracing, and metrics; set the function timeout below the calling system's timeout.

Google Cloud Run / similar

• Containers with concurrency per instance (--concurrency), --min-instances to reduce cold starts, and CPU boost for startup.
• CPU is throttled outside requests unless "CPU always allocated" is set; background tasks need always-on CPU or a separate worker.
• Configure startup probes and load balancer timeouts to match app graceful shutdown.
• Use the platform identity (service account/workload identity) for secrets and cloud access instead of static keys.

Background Jobs

| Tool | Backend | Best for | |---|---|---| | Celery | Redis/RabbitMQ/SQS | Large ecosystems, routing, schedules, mature ops | | ARQ | Redis | Async-first small/medium services | | Dramatiq | Redis/RabbitMQ | Simpler API, good throughput, middleware | | RQ | Redis | Minimal setups, straightforward jobs | | Cloud queues + workers | SQS/PubSub | Serverless-native, managed scaling |

• Idempotency key on every task: retries and at-least-once delivery are guarantees, not possibilities.
• Retries: exponential backoff with jitter, capped attempts, and a dead-letter queue you actually monitor.
• Acknowledge after success (acks_late) to avoid losing work on worker crashes; pair with idempotency because messages may be redelivered.
• Set per-task time limits and heartbeats so stuck workers are killed and retried.
• Create connections/sessions per task or per loop; never share a request-scoped AsyncSession.
• During rolling deploys, old and new workers consume the same queue: version payloads and keep handlers backward/forward compatible.
• Schedules (cron) run once per cluster, not per replica: use a scheduler/leader election.

Deployment Flow

1. CI builds the wheel/image, runs tests, and publishes the artifact with a build SHA.
2. Migration job runs against the new schema before new code starts (expand/contract safe).
3. Roll out new version gradually; readiness gates traffic.
4. Smoke tests hit /readyz and one critical read/write path.
5. Watch error rate, p99 latency, saturation, and logs for one full traffic cycle.
6. Roll back by redeploying the previous image/tag; migrations must be compatible with both versions during the window.

Anti-Patterns

• .env copied into the image or secrets in build args/layers.
• Running as root, with a shell and package manager in the runtime image for convenience.
• latest tags and mutable image references in production.
• Migrations in every replica's entrypoint, racing each other.
• No graceful shutdown: SIGTERM kills requests and loses queued work.
• Liveness probes that hit the database; readiness probes that do heavy work.
• One worker with a huge memory limit instead of horizontal replicas.
• Unlimited Celery/ARQ retries without backoff or DLQ; duplicate side effects.
• Logging to files or stdout without flush; using print for structured events.
• Scaling connection pools with replicas without a pooler.

Deployment Checklist

• [ ] Image built from a locked build, pinned base, non-root, no secrets, small surface.
• [ ] Runtime env vars set (unbuffered, fault handler, malloc arenas, UTC).
• [ ] Process model and worker/replica counts justified by measurement.
• [ ] Graceful shutdown implemented in lifespan and bounded by server timeout.
• [ ] terminationGracePeriodSeconds/LB drain aligned with app shutdown.
• [ ] /healthz and /readyz implemented, cheap, and wired to the right probes.
• [ ] Migrations run as a separate ordered job; old code survives the new schema.
• [ ] Resource requests/limits set; pools sized against DB max connections.
• [ ] Serverless-specific cold start, connection, and timeout controls in place.
• [ ] Background tasks idempotent, retried with backoff, dead-lettered and monitored.
• [ ] Rollback path tested with the previous artifact.
• [ ] Post-deploy smoke test and alerting window defined.