From production systems to offer letters.

x = 10
print(type(x), id(x))      # <class 'int'>  140...
print(isinstance(x, int))  # True — prefer isinstance over type() ==

# Truthiness: empty containers / 0 / None / "" are falsy
if not []:        print("empty list is falsy")
if (0 or "fallback"): print("or returns first truthy → 'fallback'")

Hold this picture: an object lives on the heap and owns three things forever — an identity (its address, via id()), a type (fixed at creation), and a value. A name is a label in a namespace dict that points at an object. Assignment never copies a value; it only re-points a label. That single rule explains aliasing, garbage collection timing, and why two names can mutate "each other".

Python is dynamically typed (the type check happens at runtime, on the object) yet strongly typed (it will not silently coerce "3" + 5). Mixing those up is a classic interview slip. Type hints add an optional, erased layer — they are read by tools like mypy but the interpreter ignores them at runtime.

# strong typing: no implicit string/number coercion
try:
    total = "3" + 5            # TypeError, unlike JS/PHP
except TypeError as e:
    print("refused:", e)

# duck typing: behaviour, not declared type, decides usability
def total_len(items):
    return sum(len(x) for x in items)   # works for list/str/tuple/dict

print(total_len(["ab", "cde"]))   # 5
print(total_len(("x", "yz")))     # 3 — same code, different type

# type hints are erased at runtime — they don't enforce anything
def greet(name: str) -> str:
    return "hi " + name
print(greet(123) if False else greet("Sam"))  # hint is advisory only

# BUG: default list is created once, shared across calls
def add(item, bucket=[]):   # ❌
    bucket.append(item); return bucket
add(1); add(2)            # → [1, 2]  (leaks between calls!)

def add(item, bucket=None):  # ✅ sentinel pattern
    if bucket is None: bucket = []
    bucket.append(item); return bucket

# is vs == : identity vs equality
a = [1,2]; b = a[:]
print(a == b, a is b)   # True False — equal value, different object

Immutability is not just a restriction — it is what makes an object usable as a dict key or set member, because the hash must stay constant for the object's lifetime. It also makes objects safe to share freely (no defensive copying) and lets CPython cache/intern them. The mental rule: if you would not want a value to change under you while it sits in a set, it should be immutable.

# A tuple is immutable, but it can hold a mutable object —
# so a tuple is only hashable if ALL its members are hashable.
t = (1, [2, 3])
t[1].append(4)          # legal! the tuple slot still points to the same list
print(t)                  # (1, [2, 3, 4]) — "immutable" container, mutable content
try:
    {t: "x"}              # TypeError: unhashable type: 'list'
except TypeError as e:
    print(e)

# Freeze a coordinate so it can be a key
def freeze(point):
    return tuple(point)        # list -> tuple, now hashable

grid = {}
grid[freeze([0, 0])] = "start"
grid[freeze([1, 2])] = "goal"
print(grid[(0, 0)])      # start

import copy
config = {"retries": 3, "hosts": ["a", "b"]}
shallow = copy.copy(config)
deep    = copy.deepcopy(config)

shallow["hosts"].append("c")     # mutates the SHARED inner list
print(config["hosts"])          # ['a','b','c'] — leaked into the original!
print(deep["hosts"])            # ['a','b'] — fully independent

A closure does not copy the enclosing variables — it keeps a live reference to each captured variable through a cell object. Those cells are exposed as fn.__closure__, and the names live in fn.__code__.co_freevars. This is why two closures created from the same call share the same cell and see each other's writes.

def make_pair():
    n = 0
    def inc(): nonlocal n; n += 1; return n
    def get(): return n
    return inc, get

inc, get = make_pair()
inc(); inc()
print(get())                       # 2 — both close over the SAME n cell
print(inc.__code__.co_freevars)     # ('n',)
print(inc.__closure__[0].cell_contents)  # 2 — the live captured value

# BUG: all closures share one 'i', read AFTER the loop ends
bad = [lambda: i for i in range(3)]
print([f() for f in bad])      # [2, 2, 2]

# FIX 1: default argument snapshots the value at def-time
ok1 = [lambda i=i: i for i in range(3)]
print([f() for f in ok1])      # [0, 1, 2]

# FIX 2: a factory gives each closure its own scope
def make(i): return lambda: i
ok2 = [make(i) for i in range(3)]
print([f() for f in ok2])      # [0, 1, 2]

A decorator runs in two distinct phases that trip people up. At definition time (when Python reads the @deco line) the decorator is called once with the function and returns a replacement that gets bound to the name. At call time (every invocation) it is the wrapper that runs, deciding whether/how to delegate to the original. Decorators with arguments add a third outer layer that returns the actual decorator.

import functools, time, logging
log = logging.getLogger("perf")

def timed(_fn=None, *, threshold=0.0):
    # supports both @timed and @timed(threshold=0.5)
    def deco(fn):
        @functools.wraps(fn)
        def wrap(*a, **kw):
            t0 = time.perf_counter()
            try:
                return fn(*a, **kw)
            finally:
                dt = time.perf_counter() - t0
                if dt >= threshold:
                    log.info("%s took %.3fs", fn.__name__, dt)
        return wrap
    return deco if _fn is None else deco(_fn)

@timed(threshold=0.5)
def extract(url): time.sleep(0.6); return "ok"

extract("https://reg/api")   # logs: extract took 0.6xx s

A list comprehension is not just shorter — it is faster than an equivalent for+append loop because CPython uses a specialised LIST_APPEND opcode and skips repeated attribute lookups of .append. The decision tree: need the whole collection now → comprehension; iterate once over huge data → generator expression (lazy, O(1) memory); building a lookup → dict/set comprehension.

from collections import Counter

def rows(path):
    with open(path) as f:
        for line in f:
            yield line.rstrip().split(",")

# generator pipeline — nothing is materialised until consumed
recs    = rows("trials.csv")
phases  = (r[2] for r in recs if r[2])   # lazy filter+map
counts  = Counter(phases)                       # consumes the stream once

for phase, n in counts.most_common(3):
    print(f"{phase}: {n}")

# nested comp: flatten a list of lists (read left-to-right as nested for)
matrix = [[1, 2], [3, 4]]
flat = [x for row in matrix for x in row]   # [1,2,3,4]

# these two are equivalent — lambda is just sugar for a one-expression def
add = lambda a, b: a + b
def add(a, b): return a + b

# a lambda body is ONE expression — no statements allowed:
#   lambda x: x += 1        -> SyntaxError (assignment is a statement)
#   lambda x: print(x); x   -> SyntaxError (two statements)
#   lambda x: return x      -> SyntaxError (return is a statement)
ok = lambda x: x if x > 0 else 0   # conditional EXPRESSION is fine

# 1) the key= argument — by far the most common real use
trials.sort(key=lambda t: t["enrollment"])           # sort by one field
top = max(sites, key=lambda s: s.recruited)            # pick by a derived value
rows = sorted(rows, key=lambda r: (r.country, -r.n)) # multi-key sort

# 2) tiny inline transforms / callbacks
names = list(map(lambda s: s.strip().lower(), raw))     # (a comprehension is often clearer)
df["band"] = df.apply(lambda r: "big" if r.n > 100 else "small", axis=1)

# 3) a default factory that needs an argument-free callable
from collections import defaultdict
counts = defaultdict(lambda: 0)                       # or just int; lambda shines for non-trivial defaults
groups = defaultdict(lambda: {"n": 0, "ids": []})

Python's signature grammar is richer than most people use. The full order is positional-only (before /), then normal, then *args, then keyword-only (after *), then **kwargs. The / marker (PEP 570) lets you forbid passing an argument by name — useful when a parameter name is an implementation detail you may rename.

# pos-only before /, keyword-only after *
def connect(host, /, port=443, *, timeout=30, **driver_opts):
    return (host, port, timeout, driver_opts)

connect("db.local", 5432, timeout=5, ssl=True)
# host is positional-only: connect(host="x") would raise TypeError
# timeout is keyword-only: must be named, never positional

# A transparent wrapper forwards everything unchanged
def traced(fn):
    def wrap(*args, **kwargs):
        print("call", fn.__name__, args, kwargs)
        return fn(*args, **kwargs)   # re-unpack: pass through intact
    return wrap

defaults = {"retries": 3, "timeout": 30}
override = {"timeout": 5}

# later keys win — clean layered config without mutation
final = {**defaults, **override}     # {'retries':3,'timeout':5}

def run(*sources, **cfg):
    return sources, cfg

print(run(*["a", "b"], **final))   # (('a','b'), {'retries':3,'timeout':5})

With multiple inheritance Python needs one deterministic order in which to search bases. It uses C3 linearisation: the MRO of a class is the class itself, followed by a merge of the MROs of its parents and the list of parents — preserving each parent's order and never placing a class before its subclass. If no consistent order exists, the class statement itself raises TypeError. super() walks this list, which is what makes cooperative multiple inheritance work.

class A:
    def load(self): print("A")
class B(A):
    def load(self): print("B"); super().load()
class C(A):
    def load(self): print("C"); super().load()
class D(B, C):                  # the diamond
    def load(self): print("D"); super().load()

D().load()                     # D B C A  — each runs once, in MRO order
print([c.__name__ for c in D.__mro__])  # ['D','B','C','A','object']

class Money:
    __slots__ = ("cents",)              # no __dict__: less memory, fixed attrs
    def __init__(self, cents): self.cents = cents
    def __repr__(self): return f"Money({self.cents})"
    def __add__(self, o): return Money(self.cents + o.cents)  # enables +
    def __eq__(self, o): return self.cents == o.cents          # enables ==
    def __hash__(self): return hash(self.cents)             # keep hashable after __eq__
    def __lt__(self, o): return self.cents < o.cents           # enables sort/<

print(Money(150) + Money(50))           # Money(200)
print(sorted([Money(9), Money(1)]))      # [Money(1), Money(9)]

from abc import ABC, abstractmethod

class Extractor(ABC):                 # abstraction: defines the contract
    def __init__(self, name):
        self._name = name              # encapsulation: protected by convention
    @property
    def name(self): return self._name   # controlled access
    @abstractmethod
    def parse(self, raw): ...          # subclasses MUST implement

class CtgovExtractor(Extractor):       # inheritance: is-a Extractor
    def parse(self, raw): return {"phase": raw["phase"]}   # polymorphism: overrides parse

def run(ex: Extractor, raw): return ex.parse(raw)   # works for ANY Extractor

Both express abstraction, but they answer "what counts as a valid type?" differently. An abc.ABC is nominal: a class is an Extractor only if it explicitly subclasses it. A typing.Protocol is structural (static duck typing): anything with the right methods satisfies it, no inheritance required — the type checker verifies the shape.

from typing import Protocol, runtime_checkable

@runtime_checkable
class Parser(Protocol):              # a shape, not an ancestor
    def parse(self, raw: dict) -> dict: ...

class Ctgov:                          # note: does NOT subclass Parser
    def parse(self, raw): return {"phase": raw["phase"]}

class Euct:
    def parse(self, raw): return {"phase": raw.get("trialPhase")}

def run_all(parsers: list[Parser], raw: dict):
    return [p.parse(raw) for p in parsers]   # polymorphism by shape

rows = run_all([Ctgov(), Euct()], {"phase": "III", "trialPhase": "III"})
print(isinstance(Ctgov(), Parser))   # True — runtime_checkable checks methods

class Base:
    def log(self): print("Base")
class Audited(Base):
    def log(self): print("Audited"); super().log()
class Cached(Base):
    def log(self): print("Cached"); super().log()
class Service(Audited, Cached): pass

Service().log()                 # Audited -> Cached -> Base (each super() walks the MRO once)
print([c.__name__ for c in Service.__mro__])
# ['Service', 'Audited', 'Cached', 'Base', 'object']

class Account:
    def __init__(self, balance):
        self._balance = balance          # note the underscore: real storage

    @property                          # the getter — read obj.balance
    def balance(self):
        return self._balance

    @balance.setter                   # validate on write: obj.balance = 50
    def balance(self, value):
        if value < 0:
            raise ValueError("balance cannot be negative")
        self._balance = value

    @balance.deleter                  # del obj.balance
    def balance(self):
        del self._balance

    @property                          # read-only computed value — no setter
    def is_overdrawn(self):
        return self._balance < 0

a = Account(100)
a.balance = 50        # goes through the setter (validated)
a.is_overdrawn       # computed on access; assigning to it raises AttributeError

from functools import cached_property, lru_cache

class Dataset:
    @cached_property          # stored in self.__dict__; per-instance; recompute = del obj.stats
    def stats(self):
        return expensive_scan(self.path)

    @property                 # DON'T stack @property over @lru_cache:
    @lru_cache                # lru_cache keys on `self`, so it pins every
    def bad(self):           # instance alive forever -> memory leak
        return expensive_scan(self.path)

A for loop is sugar. The interpreter calls iter(obj) once to get an iterator, then calls next() on it repeatedly until StopIteration is raised — that exception is the loop's stop signal, not an error. A generator function builds this iterator for you: each yield hands back one value and freezes the frame (locals, instruction pointer); the next next() thaws it and resumes on the line after the yield.

def countdown(n):
    while n > 0:
        yield n        # pause here, return n, remember n and the line
        n -= 1         # resumes HERE on the next next()

g = countdown(3)        # nothing runs yet — calling a gen fn returns a generator
print(next(g))         # 3   (runs to the first yield)
print(next(g))         # 2
print(next(g))         # 1
next(g)                # raises StopIteration -> a for-loop catches this and stops

def running_avg():
    total = count = 0
    avg = None
    while True:
        x = yield avg          # yield is also an expression: receives send()'d values
        total += x; count += 1
        avg = total / count

a = running_avg()
next(a)                      # prime it: advance to the first yield
print(a.send(10))            # 10.0
print(a.send(20))            # 15.0  (state persists between calls)
a.close()                   # raises GeneratorExit inside, stops the coroutine

def flatten(nested):
    for sub in nested:
        yield from sub          # delegate: re-yield every item, transparently
print(list(flatten([[1, 2], [3]])))   # [1, 2, 3]

A with expr as v: block is shorthand for calling two dunders around the body. __enter__ runs first and its return value is bound to v; the body runs; then __exit__(exc_type, exc, tb) runs no matter what — normal exit passes three Nones, an exception passes its details. The teardown is guaranteed even on return, break, or a raised exception mid-body.

import time

class Timed:
    def __init__(self, label): self.label = label
    def __enter__(self):
        self.t0 = time.perf_counter()
        return self                 # bound to the 'as' target
    def __exit__(self, exc_type, exc, tb):
        dt = time.perf_counter() - self.t0
        print(f"{self.label}: {dt:.3f}s")
        if exc_type is TimeoutError:
            print("  (suppressing timeout)")
            return True            # truthy -> SWALLOW the exception
        return False                # falsy / None -> let it propagate

with Timed("query") as t:
    raise TimeoutError      # __exit__ still runs, sees it, returns True -> no crash
print("continued")              # reached, because the exception was suppressed

from contextlib import ExitStack

# open a dynamic number of resources, all closed in reverse on exit
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    merged = merge(files)        # every file guaranteed closed, even on error

# async resources need __aenter__/__aexit__, driven by 'async with'
from contextlib import asynccontextmanager
@asynccontextmanager
async def lease(pool):
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

The GIL isn't laziness — it's the price of CPython's reference counting. Every object's refcount is mutated constantly; making those increments atomic per-object would need a lock on every object and wreck single-thread speed. One global lock is the cheap alternative. A thread holds the GIL while running bytecode and releases it (a) voluntarily on blocking I/O and many C-extension calls, and (b) involuntarily every few milliseconds (the "check interval", sys.setswitchinterval) so other threads get a turn.

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu(n):                       # pure-Python CPU work: holds the GIL
    return sum(i*i for i in range(n))

def timed(executor, fn, args):
    t0 = time.perf_counter()
    with executor() as ex:
        list(ex.map(fn, args))
    return time.perf_counter() - t0

work = [5_000_000] * 4
print("threads:", timed(ThreadPoolExecutor, cpu, work))    # ~no speedup: GIL serializes
print("procs:  ", timed(ProcessPoolExecutor, cpu, work))   # ~4x: real parallelism
# swap cpu() for a requests.get() and the THREAD version wins instead

import threading
counter = 0
def bump():
    global counter
    for _ in range(1_000_000):
        counter += 1            # load, add, store: 3 bytecodes — GIL can switch between them

ts = [threading.Thread(target=bump) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
print(counter)                 # < 4_000_000: lost updates. GIL != thread safety
# fix: guard the shared state with threading.Lock()

# an object defining __get__/__set__ controls attribute access
class Positive:
    def __set_name__(self, owner, name): self.n = "_" + name
    def __get__(self, obj, owner): return getattr(obj, self.n)
    def __set__(self, obj, value):
        if value < 0: raise ValueError("must be non-negative")
        setattr(obj, self.n, value)

class Account:
    balance = Positive()        # validation runs on every assignment

class Point:
    __slots__ = ("x", "y")   # no per-instance __dict__: less memory, no new attrs

C.__mro__   # method resolution order: the exact lookup chain (C3 linearization)

Python is compiled and interpreted. Your .py is parsed to an AST, compiled to bytecode (cached in __pycache__/*.pyc), and that bytecode is run by the CPython evaluation loop — a giant dispatch over opcodes operating on a per-frame value stack. "Interpreted" means there's no machine-code build step you run; the VM executes the bytecode each time.

import dis
def add(a, b): return a + b
dis.dis(add)
# LOAD_FAST a / LOAD_FAST b / BINARY_OP + / RETURN_VALUE

# attribute lookup obj.x walks a precise chain:
# 1) data descriptor on the type (has __set__)   -> wins
# 2) instance __dict__                            -> obj.__dict__["x"]
# 3) non-data descriptor / class attr on the MRO  -> e.g. methods
# 4) __getattr__ fallback                          -> only if all miss
class C:
    cls_attr = 1
    def m(self): return 42
c = C(); c.inst_attr = 9
print(c.inst_attr, c.cls_attr, c.m())   # 9 1 42

class Singleton:
    _inst = None
    def __new__(cls):           # allocates/returns the instance (runs BEFORE __init__)
        if cls._inst is None:
            cls._inst = super().__new__(cls)
        return cls._inst
    def __init__(self): self.ready = True   # initializes the (possibly reused) instance

print(Singleton() is Singleton())     # True

class Plugin:
    registry = {}
    def __init_subclass__(cls, key, **kw):   # runs once per subclass DEFINITION
        super().__init_subclass__(**kw)
        Plugin.registry[key] = cls       # auto-register — no metaclass needed

class Ctgov(Plugin, key="ctgov"): pass
print(Plugin.registry)                # {'ctgov': <class 'Ctgov'>}

import sys
x = []                  # the list object's refcount = 1
y = x                   # 2 (another name points at it)
sys.getrefcount(x)      # 3: x, y, + the temp arg to getrefcount
del y                   # back to 1
del x                   # 0 -> object freed IMMEDIATELY (no pause)

a = {}; b = {}
a["b"] = b; b["a"] = a   # a and b reference each other
del a, b                  # names gone, but each still has refcount 1
# -> unreachable yet NOT freed by refcounting; the cyclic GC handles it

import gc
gc.collect()            # force a full collection; returns # objects freed
gc.get_threshold()      # (700, 10, 10) -> gen0, gen1, gen2 trigger ratios
gc.get_count()          # live (gen0, gen1, gen2) allocation counters
gc.disable()            # stop the CYCLIC gc (refcounting still runs)
gc.set_threshold(0)     # also disables automatic gen0 collection

# reduce footprint
__slots__                      # drop per-instance __dict__ on small objects
generators                     # stream instead of materializing big lists
numpy / array                  # packed typed buffers vs lists of boxed objects
weakref.WeakValueDictionary()  # caches that don't keep objects alive
sys.intern(s)                  # dedupe many identical strings

# find the leak
import tracemalloc
tracemalloc.start()
snap = tracemalloc.take_snapshot()
snap.statistics("lineno")[:10]   # top allocation sites
gc.get_referrers(obj)              # what still points at it?

Two systems run together. Refcounting reclaims the moment an object becomes unreachable (no pause). The generational collector is only triggered by net allocations crossing a threshold, and only scans container types (the only ones that can form cycles); ints, strings, and floats never participate. The diagram traces one object from birth to either an instant refcount-zero free or promotion through the generations.

import gc

gc.collect()                       # clean slate
print(gc.get_count())            # (gen0, gen1, gen2) live allocation counters, e.g. (12, 0, 0)
print(gc.get_threshold())        # (700, 10, 10): gen0 collects after ~700 net allocs

class Node: pass
def make_cycle():
    a, b = Node(), Node()
    a.peer = b; b.peer = a       # mutual references -> a cycle
    # a, b go out of scope here: refcount stays 1 each, NOT freed

for _ in range(5): make_cycle()
print("freed by cyclic gc:", gc.collect())   # > 0: the cycles refcounting missed

import weakref, tracemalloc

# a parent/child cycle that a normal dict would keep alive forever
class Child:
    def __init__(self, parent):
        self.parent = weakref.ref(parent)   # weak: does NOT bump parent's refcount

# a cache that releases entries when no one else holds them
cache = weakref.WeakValueDictionary()

# diff allocations over time to localize a growing leak
tracemalloc.start()
snap1 = tracemalloc.take_snapshot()
# ... run the suspect workload ...
snap2 = tracemalloc.take_snapshot()
for stat in snap2.compare_to(snap1, "lineno")[:5]:
    print(stat)              # the lines whose memory grew most -> your leak

Two checkers dominate: mypy (the reference checker) and pyright (Microsoft, powers Pylance in VS Code, very fast). They do flow-sensitive type narrowing: after if x is None: return, the checker knows x is non-None below. Hints are zero-cost at runtime — but tools like pydantic and dataclasses deliberately do read annotations to build validation and __init__. Gradual typing means you can add hints file-by-file and tighten mypy --strict over time.

from typing import Protocol, TypedDict, Optional

class SupportsClose(Protocol):       # structural / "static duck typing"
    def close(self) -> None: ...

def shutdown(res: SupportsClose) -> None:
    res.close()                       # any object WITH close() type-checks

class Trial(TypedDict):              # dict shape known to the checker
    id: str
    phase: int
    sponsor: Optional[str]            # Optional[str] == str | None

def label(t: Trial) -> str:
    s = t["sponsor"]                 # type: str | None
    if s is None:                    # narrowing: below here s is str | None
        return f("{t['id']} (unsponsored)")
    return f("{t['id']} / {s.upper()}")   # s narrowed to str — .upper() is safe

# NEW PEP 695 syntax (Python 3.12+): type params inline, no imports
def first[T](items: list[T]) -> T | None:
    return items[0] if items else None

class Box[T]:                       # generic class, no Generic[T] base
    def __init__(self, value: T) -> None:
        self.value = value

type Result[T] = T | None          # PEP 695 lazy type alias

# LEGACY (still valid, pre-3.12): explicit TypeVar + Generic
from typing import TypeVar, Generic
U = TypeVar("U")
class OldBox(Generic[U]):
    def __init__(self, value: U) -> None:
        self.value = value

print(first([1, 2, 3]))           # 1 — checker infers T = int
print(Box("hi").value)             # hi — Box[str]

All exceptions derive from BaseException; almost everything you should catch derives from Exception. Above it sit SystemExit, KeyboardInterrupt, and GeneratorExit — never swallow these with a bare except:. The four clauses split cleanly: try = risky code, except = handle a specific failure, else = ran only if no exception (keeps the try body minimal), finally = always runs, even on return or re-raise — the place for cleanup.

class TrialError(Exception):          # domain base — callers catch this
    """Base for trial-pipeline failures."""

class ParseError(TrialError):         # specific subtype
    def __init__(self, trial_id: str, reason: str):
        self.trial_id = trial_id
        super().__init__(f("{trial_id}: {reason}"))

def parse_phase(raw: dict) -> int:
    try:                              # EAFP: attempt, don't pre-check
        return int(raw["phase"])
    except (KeyError, ValueError) as e:
        # raise from: keep original cause in the traceback (__cause__)
        raise ParseError(raw.get("id", "?"), "bad phase") from e

try:
    parse_phase({"id": "T1", "phase": "x"})
except TrialError as e:           # catch the domain base → handles all subtypes
    print("handled:", e, "| cause:", repr(e.__cause__))

# Concurrent work can fail in MANY ways at once → ExceptionGroup
def run_batch():
    errors = []
    for tid in ("T1", "T2", "T3"):
        try:
            if tid != "T2":
                raise ValueError(f("{tid} invalid"))
        except Exception as e:
            errors.append(e)
    if errors:
        raise ExceptionGroup("batch failed", errors)

try:
    run_batch()
except* ValueError as eg:      # except* handles a SUBSET of the group
    print("value errors:", len(eg.exceptions))   # value errors: 2

The decorator inspects the class's __annotations__ and synthesises dunder methods. frozen=True makes instances immutable (and hashable, so they work as dict keys / set members). slots=True (3.10+) generates __slots__, cutting per-instance memory and blocking accidental new attributes. field(default_factory=list) is the correct way to give a mutable default — sharing one list across instances is the same trap as a mutable default argument. __post_init__ runs after the generated __init__ for validation or derived fields.

from dataclasses import dataclass, field

@dataclass(frozen=True, slots=True)     # immutable + memory-lean + hashable
class Trial:
    id: str
    phase: int = 1
    # default_factory: each instance gets its own list (NOT tags=[] — shared!).
    # compare=False keeps it out of __hash__ so a list field can't break hashing.
    tags: list[str] = field(default_factory=list, compare=False)
    code: str = field(init=False, default="")     # derived, not a ctor arg

    def __post_init__(self):
        if self.phase not in (1, 2, 3, 4):
            raise ValueError("phase must be 1-4")
        # frozen → must use object.__setattr__ to set derived field
        object.__setattr__(self, "code", f("{self.id}-P{self.phase}"))

t = Trial("NCT01", 3, ["oncology"])
print(t)                  # Trial(id='NCT01', phase=3, tags=['oncology'], code='NCT01-P3')
print({t})               # hashable because frozen — works in a set

from enum import Enum, IntEnum, StrEnum, auto

class Status(Enum):              # named constants; identity comparison
    DRAFT = auto()              # auto() → 1, 2, 3...
    ACTIVE = auto()
    CLOSED = auto()

class Priority(IntEnum):         # compares/sorts as ints
    LOW = 1; HIGH = 9

class Region(StrEnum):           # is-a str → JSON-friendly (3.11+)
    US = "us"; EU = "eu"

print(Status.ACTIVE, Status.ACTIVE.value)   # Status.ACTIVE 2
print(Priority.HIGH > Priority.LOW)         # True — IntEnum sorts
print(Region.EU == "eu")                   # True — StrEnum equals its str

Patterns compose: a literal matches a value; a capture (a bare lowercase name) binds whatever is there; a sequence pattern [a, b, *rest] matches and unpacks lists/tuples; a mapping pattern {"type": t} matches dict subsets (extra keys allowed); a class pattern Point(x=0, y=y) matches by type and destructures attributes. Add guards (case p if p.phase > 2) for extra conditions, | for or-patterns, and as to bind a whole sub-pattern. _ is the wildcard default.

from dataclasses import dataclass

@dataclass
class Click: x: int; y: int
@dataclass
class Key: code: str

def handle(event):
    match event:
        case Click(x=0, y=0):                 # class pattern + literal
            return "origin click"
        case Click(x=x, y=y) if x == y:        # class pattern + guard
            return f("diagonal at {x}")
        case Key(code="esc" | "q"):            # or-pattern
            return "quit"
        case {"type": "scroll", "dy": dy}:    # mapping pattern (subset)
            return f("scroll {dy}")
        case [first, *rest]:                  # sequence pattern + capture
            return f("batch of {1 + len(rest)}, head={first}")
        case _:                               # wildcard fallback
            return "unknown"

print(handle(Click(3, 3)))            # diagonal at 3
print(handle({"type": "scroll", "dy": -4}))  # scroll -4
print(handle([1, 2, 3]))             # batch of 3, head=1

OK = 200

def classify(status):
    match status:
        # WRONG: a bare OK here is read as a CAPTURE name, not the constant 200!
        # case OK: ...   ← would match EVERYTHING and rebind OK
        case 200:            # compare against a literal — fine
            return "ok"
        case http.OK:        # dotted name → treated as a VALUE to compare
            return "ok-const"
        case int() as code if code >= 500:   # class pattern + as-bind + guard
            return f("server error {code}")
        case _:
            return "other"

re.match anchors at the start only, re.search scans the whole string for the first hit, and re.fullmatch requires the pattern to consume the entire string. Most "my regex doesn't work" bugs are really "I used match when I meant search". Anchors ^/$ make intent explicit and are usually clearer than relying on which function you called.

import re

# Use a RAW string r"" so backslashes mean regex, not Python escapes.
# Named groups (?P<name>...) give you a dict instead of fragile indexes.
LOG = re.compile(
    r"(?P<ip>\d{1,3}(?:\.\d{1,3}){3})\s+"
    r"\[(?P<ts>[^\]]+)\]\s+"
    r'"(?P<method>[A-Z]+)\s+(?P<path>\S+)"\s+'
    r"(?P<status>\d{3})"
)

line = '10.0.0.7 [28/Jun/2026:10:00:00] "GET /api/users" 200'
m = LOG.search(line)
if m:
    print(m.group("method"), m.group("path"), m.group("status"))
    print(m.groupdict())   # {'ip': '10.0.0.7', 'ts': ..., 'method': 'GET', ...}

# findall returns tuples of groups; finditer yields match objects (better)
errors = [mm.group("path")
          for mm in LOG.finditer(line)
          if mm.group("status").startswith("5")]

# sub with a callback: redact emails, keep the domain
text = "reach me at sam@acme.io or jo@acme.io"
redacted = re.sub(r"[\w.]+@([\w.]+)",
                 lambda g: "***@" + g.group(1), text)
print(redacted)   # reach me at ***@acme.io or ***@acme.io

import re

# Lookaround asserts context WITHOUT consuming it (zero-width).
# Password rule: 8+ chars, at least one digit and one letter.
pw = re.compile(r"(?=.*[A-Za-z])(?=.*\d).{8,}")
print(bool(pw.fullmatch("alpha123")))   # True
print(bool(pw.fullmatch("alphabet")))   # False — no digit

# Negative lookbehind: a price NOT preceded by a currency code already.
print(re.findall(r"(?<!USD )\d+\.\d{2}", "USD 9.99 and 4.50"))  # ['4.50']

# re.VERBOSE: whitespace/comments ignored — document complex patterns.
phone = re.compile(r"""
    (\+\d{1,2}\s?)?     # optional country code
    \(?\d{3}\)?[\s.-]?  # area code
    \d{3}[\s.-]?\d{4}   # local number
""", re.VERBOSE)
print(bool(phone.search("+1 (415) 555-2671")))  # True

A naive datetime like datetime(2026, 6, 28, 10, 0) means "10:00 — somewhere, who knows". You cannot subtract a naive from an aware one (it raises), and comparing two naives from different zones silently lies. An aware datetime pins the instant. Use zoneinfo (stdlib since Python 3.9, IANA tz database) — the old pytz is no longer needed and had a famous localize() footgun.

from datetime import datetime, timezone, timedelta
from zoneinfo import ZoneInfo   # stdlib since 3.9 (IANA tz data)

# ✅ Current instant, explicitly aware in UTC.
now = datetime.now(timezone.utc)
print(now.isoformat())          # 2026-06-28T14:00:00+00:00

# A user submits a local wall-clock time in New York — attach the zone.
ny = ZoneInfo("America/New_York")
local = datetime(2026, 11, 1, 1, 30, tzinfo=ny)  # near DST fall-back

# Normalize to UTC for storage / arithmetic.
utc = local.astimezone(timezone.utc)
print(utc.isoformat())          # 2026-11-01T05:30:00+00:00

# timedelta arithmetic is unambiguous in UTC.
deadline = utc + timedelta(days=3, hours=12)
remaining = deadline - now
print(remaining.total_seconds() / 3600, "hours left")

# Display back in the user's zone only at the edge.
print(deadline.astimezone(ny).strftime("%Y-%m-%d %H:%M %Z"))

from datetime import datetime, date

# Prefer fromisoformat for machine input — fast, no format string.
dt = datetime.fromisoformat("2026-06-28T14:00:00+00:00")

# strptime when you must parse a custom human format.
human = datetime.strptime("28/06/2026 09:15", "%d/%m/%Y %H:%M")

# Unix epoch round-trip (epoch is ALWAYS UTC seconds).
ts = dt.timestamp()                         # float seconds since 1970-01-01 UTC
back = datetime.fromtimestamp(ts, tz=datetime.now().astimezone().tzinfo)

# date math: business-agnostic; .today() is naive — use .date() of an aware dt
age_days = (date(2026, 6, 28) - date(2000, 1, 1)).days
print(age_days)                            # 9675

itertools functions return iterators, not lists — they pull one item at a time, so you can chain them over a multi-gigabyte stream in constant memory. functools.lru_cache trades memory for time by memoizing pure functions. pathlib.Path overloads the / operator for joining and unifies the dozens of os.path helpers into methods — and it's cross-platform without manual separators.

from itertools import chain, groupby, islice, accumulate, pairwise
from operator import itemgetter

rows = [
    {"team": "A", "pts": 3}, {"team": "A", "pts": 5},
    {"team": "B", "pts": 2}, {"team": "B", "pts": 9},
]

# groupby needs the data PRE-SORTED on the key (it groups runs).
rows.sort(key=itemgetter("team"))
for team, grp in groupby(rows, key=itemgetter("team")):
    total = sum(r["pts"] for r in grp)
    print(team, total)              # A 8 / B 11

# islice: take a window from an infinite/large iterator without a list.
def naturals():
    n = 1
    while True:
        yield n; n += 1
print(list(islice(naturals(), 5, 10)))   # [6, 7, 8, 9, 10]

# chain flattens; accumulate runs a running total; pairwise (3.10+) windows.
print(list(chain([1, 2], [3, 4])))      # [1, 2, 3, 4]
print(list(accumulate([1, 2, 3, 4])))    # [1, 3, 6, 10]
print(list(pairwise([1, 2, 3])))         # [(1, 2), (2, 3)]

from functools import lru_cache, partial, reduce
from pathlib import Path

# lru_cache: memoize a pure, expensive function (here, recursion).
@lru_cache(maxsize=None)        # functools.cache is the 3.9+ alias for this
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)
print(fib(50))                  # instant; without cache: exponential
print(fib.cache_info())          # hits/misses/maxsize/currsize

# partial: freeze arguments to build a specialized callable.
to_int = partial(int, base=16)
print(to_int("ff"))                # 255

# reduce: fold a sequence (use sparingly — a loop is often clearer).
print(reduce(lambda a, b: a * b, range(1, 6)))   # 120 = 5!

# pathlib: build, inspect, and read paths cross-platform.
cfg = Path.home() / ".config" / "app" / "settings.toml"
print(cfg.suffix, cfg.stem, cfg.parent.name)   # .toml settings app
for py in Path(".").glob("**/*.py"):       # recursive glob
    if py.stat().st_size > 0:
        text = py.read_text(encoding="utf-8")   # one call, no open()

pyproject.toml declares a build backend (hatchling, setuptools, flit, or uv's own). A frontend (pip, build, uv) reads it, spins up an isolated build env, and asks the backend to produce a wheel (a zip you install directly) and an sdist (source tarball). You then upload to PyPI with twine or uv publish. A lockfile (uv.lock, or PEP 751 pylock.toml) pins exact transitive versions for reproducible installs.

# Create and activate an isolated environment (stdlib, no install needed).
python -m venv .venv
# Windows:  .venv\Scripts\activate
# macOS/Linux:
source .venv/bin/activate

# Install, freeze, and reproduce.
pip install "requests>=2.32" rich
pip freeze > requirements.txt        # exact pins of what's installed
pip install -r requirements.txt      # recreate elsewhere

# Editable/dev install of YOUR package (reads pyproject.toml).
pip install -e "."                    # changes to source apply live
pip install -e ".[dev]"               # with the 'dev' optional-deps group

# --- pyproject.toml ---
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "acme-tools"
version = "1.2.0"
requires-python = ">=3.10"
dependencies = ["requests>=2.32", "click>=8.1"]

[project.optional-dependencies]          # installed via .[dev]
dev = ["pytest>=8", "ruff", "mypy"]

[project.scripts]
acme = "acme_tools.cli:main"           # creates an `acme` console command

# --- build & publish (shell) ---
# pip-based:  python -m build  →  twine upload dist/*
# uv-based (fastest, Rust): one tool for venv + deps + build + publish
uv init acme-tools          # scaffold a standards-compliant project
uv add requests click       # resolve + write uv.lock + sync .venv
uv build                    # produce wheel + sdist in dist/
uv publish                  # upload to PyPI

# start a project (or `poetry init` inside an existing one)
poetry new trainhub && cd trainhub

# add deps: edits pyproject.toml, re-resolves, updates poetry.lock, installs
poetry add django celery
poetry add --group dev pytest ruff      # dev-only group, not shipped to prod
poetry remove celery

# install from the lockfile — every machine gets identical versions
poetry install                         # app deps + your package (editable)
poetry install --only main --no-root   # CI/prod: no dev deps, don't install the project itself

# re-resolve within your constraints and rewrite the lock
poetry update                          # everything; or `poetry update django`
poetry lock                            # re-lock without installing

# run things inside the managed venv (no manual activate needed)
poetry run pytest
poetry run python manage.py migrate
poetry env info --path               # where the venv lives

# ship it
poetry build                           # wheel + sdist into dist/
poetry publish -r pypi                 # upload (configure the token first)

# Poetry 2.x understands the standard [project] table (PEP 621);
# classic projects still use [tool.poetry]. Dependencies + groups:
[tool.poetry.dependencies]
python = "^3.12"
django = "^5.0"            # caret: >=5.0.0, <6.0.0  (no breaking major bump)
redis  = "~5.1"            # tilde: >=5.1.0, <5.2.0  (only patch updates)

[tool.poetry.group.dev.dependencies]
pytest = "*"
ruff   = "*"

[build-system]                # makes the project pip-installable too
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

# install the standalone binary — needs neither Python nor Rust
curl -LsSf https://astral.sh/uv/install.sh | sh        # macOS / Linux
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"     # Windows

# --- project mode: pyproject.toml + uv.lock ---
uv init myapp && cd myapp
uv add django celery        # resolve + write uv.lock + sync .venv (auto-creates it)
uv add --dev pytest ruff     # dev dependency group
uv remove celery

uv sync                     # make .venv match uv.lock EXACTLY
uv sync --frozen --no-dev    # CI/prod: don't touch the lock, skip dev deps
uv lock                     # re-resolve and rewrite uv.lock (no install)

uv run pytest               # run inside the env — auto-syncs first, no activate
uv run python manage.py migrate

# --- pip mode: a near drop-in replacement for pip + venv ---
uv venv                         # create .venv in ~10ms (vs ~1s for python -m venv)
uv pip install -r requirements.txt   # same flags as pip, 10-100x faster
uv pip install "fastapi>=0.110"
uv pip compile requirements.in --universal -o requirements.txt  # pip-tools
uv pip sync requirements.txt    # make the env match the file exactly

# --- manage Python interpreters (replaces pyenv) ---
uv python install 3.12 3.13     # download + manage multiple versions
uv python pin 3.12             # writes .python-version for the project
uv run --python 3.13 script.py

# --- run / install CLI tools (replaces pipx) ---
uvx ruff check .                  # run a tool in a throwaway env (alias for `uv tool run`)
uv tool install ruff           # install a CLI tool globally, isolated from projects

FROM python:3.12-slim
# copy the uv binary straight from Astral's published image
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
WORKDIR /app
# cache deps separately from source for fast rebuilds
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-install-project --no-dev
COPY . .
RUN uv sync --frozen --no-dev
ENV PATH="/app/.venv/bin:$PATH"
CMD ["python", "-m", "my_service"]

Get a logger per module with logging.getLogger(__name__) — that names records by their origin and forms a dotted hierarchy. Records flow up the hierarchy (propagation) to the root logger's handlers. Set levels in two places: the logger's level gates what it emits; each handler's level gates what that destination accepts. Configure handlers/formatters once at startup (ideally via dictConfig), never per call.

import logging

# One logger per module — naming by __name__ gives a clean hierarchy.
log = logging.getLogger(__name__)

# Configure ONCE at the app entry point (not in library modules).
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(name)s:%(lineno)d %(message)s",
)

def charge(user_id, cents):
    # Use %-style args, NOT f-strings: formatting is deferred
    # and skipped entirely if the level is disabled.
    log.info("charging user=%s amount=%d", user_id, cents)
    try:
        if cents < 0:
            raise ValueError("negative amount")
        return True
    except ValueError:
        # exc_info=True attaches the full traceback to the record.
        log.exception("charge failed user=%s", user_id)
        return False

charge(42, 1500)     # INFO  ... charging user=42 amount=1500
charge(42, -1)      # ERROR ... charge failed + traceback

import logging, json, sys

class JsonFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        # 'extra=' kwargs land as attributes on the record — pull them in.
        if hasattr(record, "request_id"):
            payload["request_id"] = record.request_id
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(handler)

# Attach request-scoped context via extra= (great for correlation IDs).
logging.getLogger("api").info(
    "request handled", extra={"request_id": "req-7f3a"})
# {"ts": ..., "level": "INFO", "logger": "api",
#  "msg": "request handled", "request_id": "req-7f3a"}

A dict/set is a hash table: the key is run through hash(), the low bits index into a backing array of slots, and the value lands there directly — no scan. That's why lookup is average O(1): you compute the slot, you don't search for it. The cost you pay is hashing the key and tolerating collisions (two keys → same slot), which CPython resolves by open addressing + probing. Average O(1) holds while the table stays under its load factor; it resizes (re-hashing everything) when it fills, which is why insert is "amortised" O(1), not worst-case.

Amortised O(1) means: averaged over a long run of operations, each costs O(1) — even though one occasional op is expensive. A list.append is the canonical case: usually free, but when the backing array fills it allocates a bigger one and copies everything (O(n)). Because the array grows geometrically (~1.125× in CPython, doubling in many languages), those copies are rare enough that the per-append average stays constant. Don't confuse amortised O(1) (dict insert, list append) with true worst-case O(1) (list index by position) — an adversarial input or a resize can spike a single op.

import timeit, random

n = 100_000
data = [random.randint(0, n) for _ in range(n)]
as_list = data                      # membership is O(n)
as_set  = set(data)                 # membership is O(1) avg
needle  = -1                         # worst case: not present → full scan for list

t_list = timeit.timeit(lambda: needle in as_list, number=1000)
t_set  = timeit.timeit(lambda: needle in as_set,  number=1000)
print(f"list in: {t_list:.4f}s   set in: {t_set:.6f}s")
print(f"set is ~{t_list / t_set:,.0f}x faster")   # typically 1000x+ at this n

# The O(n^2) -> O(n) refactor in one screen:
def has_dupe_slow(xs):                # O(n^2): scan seen-list each time
    seen = []
    for x in xs:
        if x in seen: return True
        seen.append(x)
    return False

def has_dupe_fast(xs):                # O(n): set membership is O(1) avg
    seen = set()
    for x in xs:
        if x in seen: return True
        seen.add(x)
    return False

namedtuple gives you a lightweight, immutable, memory-efficient record with named fields — clearer than a tuple of mystery indices, lighter than a class, and it stays hashable so it works as a dict key or set member. deque is a doubly-linked block list: O(1) push/pop at both ends (a plain list is O(n) at the front), plus a maxlen that makes a fixed-size sliding window or a "last N events" buffer trivial.

from collections import namedtuple, deque

# namedtuple: a self-documenting record, still a tuple (hashable, unpackable)
Trial = namedtuple("Trial", "gdcid phase indication")
t = Trial("GDC-91", 3, "NSCLC")
print(t.phase, t[1])          # 3 3 — name OR index
late = {t for t in [t] if t.phase >= 3}   # usable in a set: it's hashable

# deque with maxlen: a fixed sliding window that drops the oldest for free
window = deque(maxlen=3)
for x in [10, 20, 30, 40]:
    window.append(x)        # appending the 4th auto-evicts 10
print(list(window))            # [20, 30, 40]
window.appendleft(5)          # O(1) at the front (list would be O(n))

heapq turns a plain list into a binary min-heap in place: heappush/heappop are O(log n) and the smallest item is always at [0]. For a max-heap or a priority queue with a custom key, push (priority, item) tuples (negate for max). bisect does binary search on an already-sorted list — O(log n) to find the insertion point — so you can keep a list sorted as items arrive (insort) or bucket scores into grades without a chain of if.

import heapq, bisect

# priority queue: a scheduler that always pops the most urgent task
pq = []
for prio, name in [(3, "low"), (1, "urgent"), (2, "mid")]:
    heapq.heappush(pq, (prio, name))
print(heapq.heappop(pq))     # (1, 'urgent') — smallest priority first

# bisect: classify a value into ordered buckets in O(log n)
cuts  = [60, 70, 80, 90]
grade = "FDCBA"
def to_grade(score):
    return grade[bisect.bisect_right(cuts, score)]
print([to_grade(s) for s in [55, 73, 95]])   # ['F', 'C', 'A']

The screen is won at the moment of recognition, not the typing. Each pattern has a tell in the prompt; once you name it, the skeleton is muscle memory. The flow below is the triage you run in the first 60 seconds.

# TWO POINTERS — sorted array, find a pair summing to target
def pair_sum(nums, target):     # nums sorted ascending
    lo, hi = 0, len(nums) - 1
    while lo < hi:
        s = nums[lo] + nums[hi]
        if   s == target: return (lo, hi)
        elif s <  target: lo += 1     # need bigger → move left ptr up
        else:            hi -= 1     # need smaller → move right ptr down
    return None                    # O(n) time, O(1) space

# HASHING — unsorted two-sum, one pass, value -> index
def two_sum(nums, target):
    seen = {}
    for i, x in enumerate(nums):
        if target - x in seen:
            return (seen[target - x], i)
        seen[x] = i                # O(n) time, O(n) space
    return None

from collections import deque

# SLIDING WINDOW — shortest subarray with sum >= target (positives)
def min_window(nums, target):
    start = total = 0; best = float("inf")
    for end, x in enumerate(nums):
        total += x                 # grow window to the right
        while total >= target:      # shrink from the left while valid
            best = min(best, end - start + 1)
            total -= nums[start]; start += 1
    return 0 if best == float("inf") else best

# BFS — shortest hops in an unweighted graph (adjacency dict)
def bfs_dist(graph, src):
    dist = {src: 0}; q = deque([src])
    while q:
        node = q.popleft()         # FIFO = level order
        for nxt in graph[node]:
            if nxt not in dist:
                dist[nxt] = dist[node] + 1
                q.append(nxt)
    return dist

# DFS — count connected components, iterative (no recursion-depth risk)
def components(graph):
    seen = set(); count = 0
    for start in graph:
        if start in seen: continue
        count += 1; stack = [start]
        while stack:           # LIFO = depth first
            node = stack.pop()
            if node in seen: continue
            seen.add(node)
            stack.extend(graph[node])
    return count

A join is a filter over the cross-product of two tables. The join type decides what happens to unmatched rows: INNER keeps only matches; LEFT keeps all left rows and pads the right with NULL; FULL keeps everything. The classic bug is a missing or non-unique join key fanning out rows (a one-to-many becomes a row multiplier) and silently inflating a SUM.

-- LEFT JOIN: every trial, even those with no recorded sites (sites = NULL)
SELECT t.gdcid, t.phase, COUNT(s.site_id) AS n_sites
FROM trials t
LEFT JOIN sites s ON s.gdcid = t.gdcid     -- COUNT(s.site_id) ignores NULLs → 0
WHERE t.phase = 3
GROUP BY t.gdcid, t.phase
HAVING COUNT(s.site_id) = 0;            -- phase-3 trials with no sites

A window function computes across a set of rows related to the current row but, unlike GROUP BY, keeps every row. That's how you do running totals, rankings within a group, and "compare each row to its group's average" in one pass. The OVER (PARTITION BY … ORDER BY …) clause defines the window.

-- rank trials by enrollment within each phase, keep all rows
SELECT gdcid, phase, enrollment,
       RANK() OVER (PARTITION BY phase ORDER BY enrollment DESC) AS rnk,
       AVG(enrollment) OVER (PARTITION BY phase)            AS phase_avg,
       SUM(enrollment) OVER (ORDER BY start_date
                            ROWS UNBOUNDED PRECEDING)         AS running_total
FROM trials;
-- ROW_NUMBER for dedupe: keep the latest row per key
--   ROW_NUMBER() OVER (PARTITION BY gdcid ORDER BY updated_at DESC) = 1

A CTE (WITH … AS (…)) names a subquery so a complex pipeline reads top-to-bottom instead of nesting inside-out; it can also be recursive (org charts, graph reachability). EXPLAIN (and EXPLAIN ANALYZE for real timings) shows the planner's chosen access path — the words you hunt for are Seq Scan (full-table read; usually bad on a big filtered table) vs Index Scan/Seek, plus the join algorithm (nested-loop vs hash vs merge).

WITH site_counts AS (                -- aggregate the many-side ONCE
    SELECT gdcid, COUNT(*) AS n_sites
    FROM sites GROUP BY gdcid
)
SELECT t.gdcid, t.phase, sc.n_sites
FROM trials t
LEFT JOIN site_counts sc ON sc.gdcid = t.gdcid;   -- no fan-out, no inflated sums

EXPLAIN ANALYZE SELECT * FROM trials WHERE phase = 3;
-- look for: Index Scan using idx_trials_phase  (good)
--   vs:      Seq Scan on trials  Filter: (phase = 3)  (add an index)

CAP only describes behaviour during a partition. PACELC extends it: if Partition, trade Availability vs Consistency; Else (normal operation), trade Latency vs Consistency. That "Else" is what you actually feel day-to-day — a strongly-consistent store pays a latency tax on every write (quorum/round-trips) even when nothing is failing. Dynamo-style stores (Cassandra, DynamoDB) are PA/EL: available under partition, low-latency normally, at the cost of consistency. Spanner-style systems are PC/EC: consistent always, latency is the price.

"Eventually consistent" is not a single thing — the spectrum runs from strong (every read sees the latest write) through read-your-writes and monotonic reads (you never go backward in time) to plain eventual (replicas converge someday). Many stores let you tune this per query via quorums: with N replicas, choosing read+write quorums where R + W > N guarantees a read overlaps the latest write — strong consistency. Drop below that and you trade staleness for latency and availability.

-- compute is a named, resizable resource you turn on per workload
CREATE WAREHOUSE etl_wh WITH warehouse_size = 'MEDIUM'
  auto_suspend = 60 auto_resume = TRUE;   -- pause when idle = save money

-- instant dev copy of prod, no storage cost until you change it
CREATE TABLE trials_dev CLONE trials_prod;

-- query the table as it looked 2 hours ago
SELECT * FROM trials AT(offset => -7200);

-- call an LLM from SQL (Cortex) to summarise rows
SELECT snowflake.cortex.summarize(notes) FROM site_inspections;

Snowflake's architecture is three decoupled layers, and almost every interview answer reduces to which one you're talking about. Storage = your data, kept once as compressed columnar micro-partitions in cloud object storage (S3/Azure Blob/GCS). Compute = virtual warehouses, ephemeral MPP clusters you size and suspend per workload. Cloud Services = the always-on brain that does query optimisation, transaction management, security, and metadata. You pay storage and compute on separate meters — that decoupling is the whole product.

Snowflake auto-divides every table into micro-partitions of ~50–500 MB uncompressed, each storing columns separately with per-partition metadata (min/max, distinct counts, nulls). A WHERE filter is answered by partition pruning: the optimiser reads the metadata, skips partitions whose min/max can't match, and only scans the survivors — no B-tree indexes to design or maintain. Data arriving in roughly sorted order prunes well for free; when query patterns drift off the natural order you add automatic clustering on a clustering key so background re-clustering keeps pruning effective. The 2026 engine extends pruning to Iceberg tables, Top-K, and LIKE predicates.

-- keep pruning healthy on a column you filter that isn't the load order
ALTER TABLE events CLUSTER BY (event_date, registry);
SELECT system$clustering_information('events', '(event_date)');  -- check overlap depth

-- semi-structured JSON lands in a VARIANT and is queried with path + FLATTEN
SELECT v:patient:age::int AS age, f.value:dose::float
FROM raw_payloads,
     LATERAL FLATTEN(input => v:medications) f;

-- three layers of caching, no config: result cache (24h, exact-query reuse),
-- local SSD cache on the warehouse, and remote storage. Re-run = often free.

-- 2026 AISQL (Cortex) — AI_ as first-class SQL operators over columns
SELECT review_id,
       AI_CLASSIFY(body, ['bug', 'praise', 'billing']):labels[0] AS topic,
       AI_SENTIMENT(body)             AS mood,
       AI_COMPLETE('llama3.1-70b', 'Summarise in 8 words: ' || body) AS tldr
FROM reviews
WHERE created >= dateadd(day, -7, current_date());

import pandas as pd
df = pd.read_csv("trials.csv")                 # also read_excel, read_json, read_sql
df = pd.read_parquet("trials.parquet")         # columnar: faster + smaller than CSV
df.to_parquet("out.parquet", index=False)    # prefer parquet for re-use
chunks = pd.read_csv("huge.csv", chunksize=100_000)  # stream a file too big for RAM

df["phase"]                          # a column (Series)
df[["phase", "status"]]              # several columns (DataFrame)
df.loc[df["status"] == "Recruiting", "phase"]   # LABEL-based: rows by mask, one col
df.iloc[0:5, 0:3]                # POSITION-based: first 5 rows, 3 cols
df[(df.phase == "2") & (df.enrollment > 100)]   # boolean AND — wrap each term in ()
df[df.registry.isin(["NCT", "CTRI"])]         # membership filter

# filter + sort
act = df[df.status == "Recruiting"].sort_values("enrollment", ascending=False)

# groupby + aggregate (split-apply-combine)
g = df.groupby("phase").agg(
        n=("nct_id", "count"),
        avg_enroll=("enrollment", "mean"))          # named aggregations

# merge = SQL join; concat = stack
joined = df.merge(sites, on="nct_id", how="left")     # inner|left|right|outer
stacked = pd.concat([jan, feb], ignore_index=True)    # append rows

# reshape: long<->wide
wide = df.pivot_table(index="registry", columns="phase", values="enrollment", aggfunc="mean")
long = wide.melt(ignore_index=False)               # unpivot back to long

df.isna().sum()                       # count nulls per column
df.dropna(subset=["enrollment"])        # drop rows missing a key field
df["enrollment"] = df.enrollment.fillna(df.enrollment.median())  # impute
df.assign(flag=df.sponsor.isna())       # derive a column, chain-friendly

df.info(memory_usage="deep")            # see real memory
df["registry"] = df.registry.astype("category")   # repeated strings → huge RAM win
df["enrollment"] = pd.to_numeric(df.enrollment, downcast="integer")
df["start"] = pd.to_datetime(df.start)   # real datetimes, not strings

ts = df.set_index("start").sort_index()
ts.resample("M")["nct_id"].count()      # trials per month
ts["enrollment"].rolling(7).mean()       # 7-period moving average
g.groupby("phase")["enrollment"].transform("mean")  # broadcast group stat back to rows

out = (df
       .query("status == 'Recruiting'")
       .assign(yr=lambda d: pd.to_datetime(d.start).dt.year)
       .groupby("yr", as_index=False)
       .agg(n=("nct_id", "count"))
       .sort_values("yr"))               # reads top-to-bottom, no temp vars

The card already says "vectorise, not loop" — here is the decision tree underneath it. A true vectorised op runs one C loop over a NumPy buffer. .apply(axis=1) and .iterrows() call back into Python per row and build a Series each time — slowest. .apply on a Series (axis-free) is a Python-level loop too, but cheaper. .map on a Series is element-wise and accepts a dict (great for lookups). For string/date work use the accessors (.str, .dt) which are vectorised, not .apply(str.upper).

import numpy as np
# SLOW: row-wise Python callback, materialises a Series per row
df["band"] = df.apply(lambda r: "big" if r.enrollment > 100 else "small", axis=1)

# FAST: vectorised — np.where over the whole column in C
df["band"] = np.where(df.enrollment > 100, "big", "small")

# many branches: np.select beats a chain of apply()
conds  = [df.enrollment > 500, df.enrollment > 100]
labels = ["mega", "big"]
df["band"] = np.select(conds, labels, default="small")

# lookup: .map with a dict is the idiomatic recode, not apply
df["region"] = df.country.map({"US": "NA", "BR": "LATAM"}).fillna("other")

# string/date work: use the vectorised accessors, never apply()
df["sponsor"] = df.sponsor.str.strip().str.upper()
df["qtr"]     = df.start.dt.quarter

The card shows the fix; here's the why. Pandas can't always tell whether an indexing result is a view (shares the parent's NumPy buffer) or a copy. Chained indexing df[mask]["col"] = v compiles to two calls: __getitem__ returns a possibly-temporary object, then __setitem__ writes to that — which may be a copy that's discarded, so your write vanishes. The single-call form df.loc[mask, "col"] = v goes through one __setitem__ on the original, which is unambiguous. The other reliable cause is operating on a slice you forgot to .copy(). Pandas 3.0's Copy-on-Write (default) ends the ambiguity: every indexing result behaves like an independent object, the warning is retired, and you opt into mutation explicitly with .copy() or .loc.

# transform: returns a result aligned to the ORIGINAL index (same length)
df["z"] = (df.enrollment - df.groupby("phase").enrollment.transform("mean")) \
        / df.groupby("phase").enrollment.transform("std")   # per-group z-score

# filter: keep whole GROUPS by a group-level predicate (not rows)
big = df.groupby("sponsor").filter(lambda g: len(g) >= 10)

# validate a merge so a silent fan-out can't happen
j = df.merge(sites, on="nct_id", how="left",
             validate="m:1",          # raise if sites isn't unique on key
             indicator=True)            # _merge col: left_only / both — audit match rate
assert (j._merge == "both").mean() > 0.95      # >95% matched or investigate

import numpy as np
np.array([[1, 2], [3, 4]])      # from a list
np.zeros((2, 3)); np.ones((2, 3)); np.full((2, 2), 9)
np.arange(0, 10, 2)               # [0 2 4 6 8]
np.linspace(0, 1, 5)               # 5 evenly spaced points
np.eye(3)                            # identity matrix

a = np.arange(12)
a.reshape(3, 4)        # 3x4  (use -1 to infer: reshape(3, -1))
a.reshape(3, 4).T       # transpose to 4x3 (just swaps strides)
a.ravel()                # flatten to 1-D (a view when possible)
a[:, np.newaxis]         # add an axis: column vector

a = np.arange(10)
a[2:7:2]                  # slice start:stop:step
a[[0, 3, 9]]                 # fancy: pick indices
a[a > 5]                     # boolean mask: elements over 5
a[a > 5] = 0                 # assign through a mask
np.where(a > 5, 1, 0)         # vectorized if/else

# shapes align from the RIGHT; a dim of 1 stretches to match
A = np.ones((3, 4))           # (3, 4)
b = np.array([1, 2, 3, 4])       # (4,)   stretches across rows
A + b                            # each row gets b added, no loop
col = np.array([[10], [20], [30]])  # (3, 1) stretches across columns
A + col                          # each column gets col added

a = np.arange(6).reshape(2, 3)
np.exp(a); np.sqrt(a); a ** 2      # element-wise, vectorized
a.sum()                          # 15  (everything)
a.sum(axis=0)                    # down columns: shape (3,)
a.sum(axis=1)                    # across rows:  shape (2,)
a.mean(); a.std(); a.max(axis=1)

The card states the rule; here is the exact procedure NumPy runs. (1) Right-align the two shapes and left-pad the shorter with 1s. (2) For each dimension, sizes are compatible if they're equal or one is 1; otherwise raise. (3) The output dimension is the max of the two; a size-1 dimension is virtually stretched by setting its stride to 0 — no data is copied, the same element is re-read. That zero-stride trick is why (1000,1) + (1,1000) produces a million-element result without a million-element temporary for either input.

import numpy as np
a = np.arange(3).reshape(3, 1)   # (3,1)
b = np.arange(4)                  # (4,)  -> padded to (1,4)
(a + b).shape                       # (3,4) outer-style grid, no python loop

# classic use: pairwise distances without a loop
pts = np.random.default_rng(0).normal(size=(5, 2))
diff = pts[:, None, :] - pts[None, :, :]   # (5,1,2)-(1,5,2) => (5,5,2)
dist = np.sqrt((diff ** 2).sum(axis=-1))    # (5,5) distance matrix

# trap: (n,) and (n,1) do NOT broadcast the way you expect
x = np.arange(3)            # (3,)
(x + x[:, None]).shape   # (3,3) outer sum, not element-wise! keep ranks explicit

An ndarray is a view over a flat buffer plus three pieces of metadata: shape, dtype, and strides (bytes to step per axis). .T, basic slices, and reshape (when possible) just synthesise new strides over the same buffer — O(1), zero copy. .reshape must copy only when the requested layout isn't expressible as strides on the current buffer (e.g. reshaping a non-contiguous transpose); .ravel() returns a view if it can, .flatten() always copies. C-order (row-major, default) stores rows contiguously; F-order (column-major) stores columns contiguously — and the difference decides cache performance: summing along the contiguous axis is dramatically faster.

a = np.arange(12).reshape(3, 4)
a.strides           # (32, 8) on int64: 32B per row, 8B per col
a.T.strides         # (8, 32) — transpose just swaps strides, no copy
a.flags["C_CONTIGUOUS"], a.T.flags["C_CONTIGUOUS"]   # True, False

# contiguity drives speed: same data, ~order-of-magnitude gap on big arrays
big = np.random.default_rng(0).normal(size=(4000, 4000))
big.sum(axis=1)   # fast: walks contiguous rows (C-order)
big.sum(axis=0)   # slower: strides across rows, cache-unfriendly

# a stride trick: sliding windows with ZERO copy (use the safe helper)
from numpy.lib.stride_tricks import sliding_window_view
w = sliding_window_view(np.arange(6), 3)   # [[0,1,2],[1,2,3],[2,3,4],[3,4,5]]

Ufuncs (np.add, np.exp, …) are compiled element-wise kernels with broadcasting, type promotion, and an out= parameter built in. np.vectorize is not vectorisation — it's a convenience wrapper around a Python loop, so it gives the API but not the speed; reach for real ufuncs, np.where/np.select, or Numba instead. Two real levers: ufunc methods like .reduce / .accumulate / .outer, and in-place ops (out= or +=) to avoid allocating a fresh array on hot paths.

a = np.arange(1, 5, dtype=np.float64)
np.add.reduce(a)            # 10.0  (same as a.sum, the reduction form)
np.multiply.outer([1,2], [3,4])  # outer product via ufunc method

# in-place: no new allocation, writes back into a (mind the dtype!)
np.exp(a, out=a)            # a is overwritten with exp(a)
a /= a.sum()              # normalise in place

from sqlalchemy import create_engine, select, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass
class Trial(Base):
    __tablename__ = "trials"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

engine = create_engine("postgresql+psycopg://...")
with Session(engine) as s:
    rows = s.scalars(select(Trial).where(Trial.id == 42)).all()
    s.add(Trial(title="new")); s.commit()      # unit of work: flush + commit

A Session is a workspace plus a transaction. Every object it loads or adds is tracked in its identity map — a dict keyed by class+primary-key — so two queries for the same row return the same Python object, and the session knows exactly what changed. Objects move through states: transient (new, not added) → pending (added, not yet in DB) → persistent (flushed/loaded, has a PK, tracked) → detached (session closed) / deleted. A flush emits the pending INSERT/UPDATE/DELETE SQL in dependency order inside the transaction; commit flushes then commits. Flush ≠ commit — autoflush sends SQL before a query so your reads see your own writes, but nothing is durable until commit.

from sqlalchemy import ForeignKey, select, func
from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column,
                            relationship, Session, selectinload, joinedload)

class Base(DeclarativeBase): pass

class Sponsor(Base):
    __tablename__ = "sponsors"
    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    # lazy="raise" makes any accidental lazy load blow up loudly in dev/tests
    trials: Mapped[list["Trial"]] = relationship(back_populates="sponsor", lazy="raise")

class Trial(Base):
    __tablename__ = "trials"
    id:         Mapped[int] = mapped_column(primary_key=True)
    title:      Mapped[str]
    sponsor_id: Mapped[int] = mapped_column(ForeignKey("sponsors.id"))
    sponsor:    Mapped[Sponsor] = relationship(back_populates="trials")

with Session(engine) as s:
    # N+1 FIX: one query for sponsors + one batched IN() query for all trials
    stmt = select(Sponsor).options(selectinload(Sponsor.trials))
    for sp in s.scalars(stmt):
        print(sp.name, len(sp.trials))      # no extra query per sponsor

    # many-to-one: joinedload pulls parent+child in ONE join
    t = s.scalars(select(Trial).options(joinedload(Trial.sponsor))).first()

    # aggregate in SQL, not Python (2.0 select + func)
    counts = s.execute(
        select(Trial.sponsor_id, func.count())
            .group_by(Trial.sponsor_id)).all()

# one-time
alembic init migrations
# point env.py target_metadata = Base.metadata, then autogenerate a diff
alembic revision --autogenerate -m "add sponsor.country"
alembic upgrade head        # apply; alembic downgrade -1 to roll back one
alembic current; alembic history --verbose

A singly linked list stores only a forward pointer; a doubly linked list adds a prev pointer so you can delete a node in O(1) without first walking to its predecessor — the reason collections.deque and an LRU cache use one internally. The cost is an extra pointer per node and two links to fix on every edit. A sentinel/dummy head node (a fake node before the real first one) removes almost all "is this the head?" special-casing — senior code uses one by default.

class Node:
    def __init__(self, val, nxt=None):
        self.val, self.next = val, nxt

class LinkedList:
    def __init__(self):
        self.head = None

    def push_front(self, val):          # O(1) insert at head
        self.head = Node(val, self.head)

    def delete(self, target):           # O(n) find, O(1) unlink
        dummy = Node(None, self.head)  # sentinel kills head edge-case
        prev, cur = dummy, self.head
        while cur:
            if cur.val == target:
                prev.next = cur.next       # skip the node = delete
                break
            prev, cur = cur, cur.next
        self.head = dummy.next

    def reverse(self):                  # O(n) time, O(1) space
        prev, cur = None, self.head
        while cur:
            cur.next, prev, cur = prev, cur, cur.next  # flip one link
        self.head = prev

    def __iter__(self):
        cur = self.head
        while cur:
            yield cur.val
            cur = cur.next

ll = LinkedList()
for x in [3, 2, 1]: ll.push_front(x)
ll.reverse()
print(list(ll))               # [3, 2, 1]

Many list problems collapse if you walk two pointers at different speeds. Move fast two steps and slow one: when fast hits the end, slow sits at the middle (one pass, no length needed). The same trick finds the k-th from the end (start fast k nodes ahead) and detects a cycle (Floyd's tortoise & hare — if there is a loop the two pointers must eventually meet).

def has_cycle(head):              # Floyd's algorithm, O(1) space
    slow = fast = head
    while fast and fast.next:
        slow = slow.next            # 1 hop
        fast = fast.next.next       # 2 hops
        if slow is fast:           # they collided → loop exists
            return True
    return False

A Python list appends in amortised O(1) but list.pop(0) is O(n) — it shifts every remaining element left. A deque is a doubly linked list of fixed-size blocks, so append/appendleft/pop/popleft are all O(1). Use a list for a stack (pop from the end is O(1)); use a deque the moment you touch the front. For thread-safe producer/consumer hand-off, reach for queue.Queue instead — it adds locking and blocking.

def is_balanced(s):
    pairs = {')': '(', ']': '[', '}': '{'}
    stack = []
    for ch in s:
        if ch in '([{':
            stack.append(ch)                 # open → push
        elif ch in pairs:
            if not stack or stack.pop() != pairs[ch]:
                return False             # mismatch or nothing to close
    return not stack                     # all opens were closed

print(is_balanced("a(b[c]{d})"))    # True
print(is_balanced("([)]"))           # False — wrong nesting

from collections import deque

def bfs(graph, start):                 # level-order, shortest hops
    seen, q, order = {start}, deque([start]), []
    while q:
        node = q.popleft()             # FIFO → explore nearest first
        order.append(node)
        for nb in graph[node]:
            if nb not in seen:
                seen.add(nb)
                q.append(nb)
    return order

def next_greater(nums):              # monotonic decreasing stack, O(n)
    res, stack = [-1] * len(nums), []   # stack holds indices
    for i, x in enumerate(nums):
        while stack and nums[stack[-1]] < x:
            res[stack.pop()] = x       # x is the next-greater for that index
        stack.append(i)
    return res

print(next_greater([2, 1, 3, 0]))   # [3, 3, -1, -1]