Python cheatsheet

nums.sort()                              # in place, returns None!sorted(nums)                             # new list, works on any iterablesorted(nums, reverse=True)sorted(words, key=len)                   # by lengthsorted(words, key=str.lower)             # case-insensitivesorted(users, key=lambda u: u.age)       # by attributesorted(users, key=lambda u: (u.dept, -u.salary))   # multi-key: dept asc, salary descfrom operator import itemgetter, attrgettersorted(rows, key=itemgetter(2))          # by column 2sorted(rows, key=itemgetter("date"))     # list of dictssorted(users, key=attrgetter("name"))# Descending on a non-numeric secondary key: sort twice (Timsort is stable)data.sort(key=itemgetter("name"))            # secondary firstdata.sort(key=itemgetter("dept"), reverse=True)  # primary lastmax(nums), min(words, key=len)           # min/max take key= toosorted(d.items(), key=itemgetter(1), reverse=True)   # dict by valueimport bisect                            # keep a list sorted / binary searchi = bisect.bisect_left(sorted_list, x)   # insertion point (== index if present)bisect.insort(sorted_list, x)import heapq                             # top-k without full sort: O(n log k)heapq.nlargest(3, nums)heapq.nsmallest(3, users, key=lambda u: u.age)

lst[-1], lst[-2]                         # last, second to lastlst[2:5], lst[:3], lst[3:], lst[::2]     # slices (copy)lst[::-1]                                # reversed copylst[:]  /  lst.copy()                    # shallow copylst[1:3] = [10, 20, 30]                  # splice: replace a slicelst.append(x)        # one element         O(1)lst.extend(iter)     # many elements       lst += iterlst.insert(0, x)     # O(n) — use deque for queueslst.pop()            # last, O(1)lst.pop(0)           # first, O(n) — again: dequelst.remove(x)        # first occurrence by VALUE, ValueError if absentdel lst[i]           # by index[x * 2 for x in nums if x > 0]           # filter + map in one[a if a else "-" for a in cells]         # ternary inside: transform all[y for row in grid for y in row]         # flatten one level (left-to-right nesting)list(dict.fromkeys(items))               # dedupe KEEPING orderlist(set(items))                         # dedupe, order lost", ".join(str(x) for x in nums)          # join non-stringsx in lst                                 # O(n) — convert to set if repeatedlst.index(x)                             # first index, ValueError if absentlst.count(x)# init pitfallsgrid = [[0] * 4 for _ in range(3)]       # RIGHTgrid = [[0] * 4] * 3                     # WRONG: 3 refs to the SAME row

d = {"a": 1, "b": 2}d.get(k)                                 # None if missing (no raise)d.get(k, default)d.setdefault(k, []).append(x)            # init-if-missing, then used.pop(k, None)                           # remove, no raisemerged = d1 | d2                         # 3.9+ (d2 wins); in-place: d1 |= d2{**d1, **d2, "extra": 1}                 # merge + add{k: v * 2 for k, v in d.items()}         # dict comprehension{v: k for k, v in d.items()}             # invert{k: v for k, v in d.items() if v > 0}    # filterfor k, v in d.items(): ...               # ALWAYS items() for bothmax(d, key=d.get)                        # key with max valuedict(sorted(d.items()))                  # sorted by keydict(sorted(d.items(), key=lambda kv: kv[1]))   # sorted by valuefrom collections import defaultdict, Countergroups = defaultdict(list)for u in users:    groups[u.dept].append(u)             # no key-exists check everc = Counter("abracadabra")               # {'a': 5, 'b': 2, ...}c.most_common(2)                         # [('a', 5), ('b', 2)]c["z"]                                   # 0, never KeyErrorCounter(a) - Counter(b), c1 + c2         # multiset opsnext(iter(d)), next(iter(d.values()))    # first key / value without pop

s = {1, 2, 3};  empty = set()            # {} is an empty DICTs.add(x);  s.discard(x)                  # discard: no raise if absents | t,  s & t,  s - t,  s ^ t            # union, intersection, diff, sym-diffs <= t,  s < t                           # subset, proper subsets.isdisjoint(t){x.id for x in items}                    # set comprehensionseen = set()                             # classic dedupe-while-loopingfor x in stream:    if x.key in seen: continue    seen.add(x.key)

s.split()                    # on ANY whitespace, no empty stringss.split(",", maxsplit=1)     # "k=v" style: k, _, v = s.partition("=")s.splitlines()s.strip(), s.strip(".,!")    # both ends; lstrip/rstrips.removeprefix("id_"), s.removesuffix(".txt")     # 3.9+s.replace(old, new)s.startswith(("http://", "https://"))    # tuple = ORs.lower(), s.casefold()                  # casefold for comparisonss.zfill(5), s.rjust(10), s.ljust(10, "-"), s.center(20)"".join(reversed(s)), s[::-1]            # reverses.isdigit(), s.isalpha(), s.isalnum(), s.isidentifier()"sep".join(parts)                        # NEVER += in a loop# f-stringsf"{x!r}"                     # reprf"{x=}"                      # debug: prints  x=42f"{n:,}"  f"{n:_}"           # 1,234,567 / 1_234_567f"{ratio:.1%}"               # 12.3%f"{x:.2f}"  f"{x:8.2f}"      # 2 decimals / width 8f"{n:05d}"  f"{n:#x}"  f"{n:b}"          # 00042 / 0x2a / binaryf"{s:>10}" f"{s:<10}" f"{s:^10}"         # align right/left/centerf"{dt:%Y-%m-%d %H:%M}"                   # datetime inlinef"{'a' if cond else 'b'}"                # expressions allowed

a, b = b, a                              # swapa, b, *rest = [1, 2, 3, 4, 5]            # rest = [3, 4, 5]first, *_, last = seq(a, b), c = pair, x                      # nesteddef f(*args, **kwargs): ...f(*lst, **d)                             # unpack into call[*a, *b],  (*a, *b),  {**d1, **d2}       # merge literalsfor i, x in enumerate(items, start=1): ...for a, b in zip(xs, ys): ...for a, b in zip(xs, ys, strict=True): ...   # 3.10+: raise on length mismatchlist(zip(*matrix))                       # transposedict(zip(keys, values))x, = single_item_list                    # or [x] = ...; raises if != 1 element

if (n := len(data)) > 10:                # use a value you also test    print(f"too long: {n}")while (chunk := f.read(8192)):           # read-until-empty loops    process(chunk)if (m := re.search(r"\d+", s)):          # regex match-and-use    print(m.group())[y for x in data if (y := expensive(x)) is not None]   # compute once in comp

x = a if cond else b                     # ternaryx = val or default                       # careful: 0/""/[] also fall throughx = val if val is not None else default  # the safe version0 <= i < len(lst)                        # chained comparisonif x in ("a", "b", "c"): ...             # membership instead of or-chainsfor x in reversed(lst): ...for i in range(len(lst) - 1, -1, -1): ...   # index-based reversefor x in items:    if match(x):        found = x        breakelse:                                    # for-else: runs iff NO break    raise LookupError("not found")# but usually cleaner:found = next((x for x in items if match(x)), None)any(x > 0 for x in nums)                 # short-circuitsall(u.active for u in users)             # True on empty!sum(1 for x in items if pred(x))         # count matching

from collections import deque, namedtupledq = deque(maxlen=100)                   # rolling window / recent-Ndq.appendleft(x); dq.pop(); dq.rotate(1) # O(1) both endsPoint = namedtuple("Point", "x y")       # or dataclass for anything richerp = Point(1, 2); p.x; x, y = pfrom enum import Enum, autoclass Status(Enum):    ACTIVE = auto()    DONE = auto()Status.ACTIVE.name                       # "ACTIVE"Status["ACTIVE"]                         # lookup by nameStatus(1)                                # lookup by valueimport itertools as itit.chain(a, b)                           # concat iterables lazilyit.islice(gen, 10)                       # first 10 of a generator (no slicing!)it.batched(data, 3)                      # 3.12+: chunks of 3it.pairwise(seq)                         # (s0,s1), (s1,s2), ...  3.10+it.groupby(sorted(rows, key=k), key=k)   # MUST sort by same key firstit.product(range(3), repeat=2)           # nested loops flattenedit.combinations(items, 2)                # pairs, no repeatsit.count(10), it.cycle("AB"), it.repeat(x, 5)from functools import reduce, cachereduce(lambda acc, x: acc | x, sets)     # fold — but sum/any/max cover 90%

from pathlib import Pathp = Path("data") / "raw" / "file.txt"    # join with /p.read_text()                            # whole file as str (encoding="utf-8")p.write_text(s)                          # atomic enough for small stuffp.exists(), p.is_file(), p.is_dir()p.name, p.stem, p.suffix, p.parent       # file.txt / file / .txt / data/rawp.with_suffix(".json")p.mkdir(parents=True, exist_ok=True)list(p.glob("*.csv")), list(p.rglob("*.py"))    # rglob = recursivePath.home(), Path.cwd(), Path(__file__).parentwith open(p) as f:                       # line-by-line, constant memory    for line in f:        line = line.rstrip("\n")import jsondata = json.loads(p.read_text())p.write_text(json.dumps(data, indent=2, default=str))   # default=str: dates etc.import csvwith open(p, newline="") as f:    for row in csv.DictReader(f): ...

q, r = divmod(17, 5)                     # (3, 2)7 // 2, -7 // 2                          # 3, -4  (floor, not truncation!)7 % 3, -7 % 3                            # 1, 2   (sign follows divisor)round(2.675, 2)                          # 2.67 — float repr, not a bug; use Decimal for moneyint("ff", 16), int("1010", 2)            # from baseabs(x), pow(2, 10), 2**10min(max(x, lo), hi)                      # clampmath.inf, -math.inf                      # initial best-so-far valuesfloat("nan") != float("nan")             # NaN never equals; use math.isnansum(xs) / len(xs), statistics.mean(xs), statistics.median(xs)x != int(x)                              # has decimal partf"{0.1 + 0.2:.17f}"                      # why == fails on floatsmath.isclose(a, b)random.choice(seq), random.sample(seq, 3), random.shuffle(lst), random.randint(1, 6)

from datetime import datetime, date, timedelta, timezonenow = datetime.now(timezone.utc)         # ALWAYS aware in serverstoday = date.today()dt = datetime.fromisoformat("2026-07-07T12:00:00+00:00")dt.isoformat()dt + timedelta(days=7, hours=3)(d2 - d1).days, (t2 - t1).total_seconds()dt.strftime("%Y-%m-%d"), datetime.strptime(s, "%d/%m/%Y")dt.timestamp(), datetime.fromtimestamp(ts, tz=timezone.utc)import timestart = time.perf_counter()              # timing code: perf_counter,elapsed = time.perf_counter() - start    # NEVER time.time()

import rem = re.search(r"(\d{4})-(\d{2})", s)     # first match anywhereif m: m.group(0), m.group(1), m.groups()re.match(...)                            # anchored at START only; fullmatch = whole stringre.findall(r"\d+", s)                    # all matches as list of stringsre.finditer(r"\d+", s)                   # lazy, gives match objectsre.sub(r"\s+", " ", s)                   # normalize whitespacere.sub(r"(\w+)@(\w+)", r"\2.\1", s)      # backrefs in replacementre.split(r"[,;]\s*", s)pattern = re.compile(r"^\w+$")           # compile if reused in a loop(?P<year>\d{4})  →  m.group("year")      # named groupsr"..." always                            # raw strings for patterns

print(f"{x=}, {y=}")                     # fastest debug printbreakpoint()                             # drops into pdb (c=continue, n=next)from pprint import pprint; pprint(obj)import sys; print(x, file=sys.stderr)isinstance(x, (int, float))              # multi-type checkcallable(f), hasattr(o, "attr"), getattr(o, "attr", default)vars(obj), dir(obj)                      # inspect anythingid_gen = iter(range(10**9))              # quick unique ids: next(id_gen)# swap keys case-insensitively, chain gets, etc.value = (d.get("a") or {}).get("b")      # nested get without KeyErrorvalue = d1.get(k) or d2.get(k) or defaultimport osos.environ.get("DEBUG", "0") == "1"os.cpu_count()import subprocessout = subprocess.run(["ls", "-l"], capture_output=True, text=True, check=True).stdoutimport argparse                          # sys.argv[1] fine for one arg;                                         # argparse the moment you have flags# quick throwaway HTTP (stdlib only)from urllib.request import urlopenbody = urlopen("https://api.example.com").read().decode()

import asyncio# Run concurrently, results in input orderresults = await asyncio.gather(*(fetch(u) for u in urls))# gather: one exception cancels NOTHING by default — it propagates the first# error but sibling tasks keep running. Use return_exceptions=True to collect.# TaskGroup (3.11+): structured concurrency — on error, cancels all siblings,# raises ExceptionGroupasync with asyncio.TaskGroup() as tg:    t1 = tg.create_task(fetch(u1))    t2 = tg.create_task(fetch(u2))result = t1.result()  # available after the block# Timeoutasync with asyncio.timeout(5):          # 3.11+, cancellation-friendly    await slow()await asyncio.wait_for(slow(), timeout=5)  # older; wraps in a Task# First completeddone, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)for t in pending:    t.cancel()await asyncio.gather(*pending, return_exceptions=True)  # await the cancels!

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutorwith ThreadPoolExecutor(max_workers=8) as pool:      # I/O-bound    results = list(pool.map(fetch_sync, urls))with ProcessPoolExecutor() as pool:                  # CPU-bound    results = list(pool.map(crunch, chunks))         # args/results must pickle

import functoolsdef timing(func):                       # plain decorator    @functools.wraps(func)              # preserves __name__, __doc__, signature    def wrapper(*args, **kwargs):        start = time.perf_counter()        try:            return func(*args, **kwargs)        finally:            print(f"{func.__name__}: {time.perf_counter() - start:.3f}s")    return wrapperdef retry(times=3, exceptions=(Exception,)):   # decorator WITH arguments:    def decorator(func):                       # factory → decorator → wrapper        @functools.wraps(func)        def wrapper(*args, **kwargs):            for attempt in range(times):                try:                    return func(*args, **kwargs)                except exceptions:                    if attempt == times - 1:                        raise        return wrapper    return decorator@retry(times=5, exceptions=(ConnectionError,))def call(): ...

# Iterator protocol: __iter__ returns self, __next__ raises StopIterationclass Countdown:    def __init__(self, n): self.n = n    def __iter__(self): return self    def __next__(self):        if self.n <= 0: raise StopIteration        self.n -= 1        return self.n + 1def chunks(seq, size):                  # generator: lazy, one-shot    for i in range(0, len(seq), size):        yield seq[i:i + size]def flatten(nested):    for item in nested:        if isinstance(item, list):            yield from flatten(item)    # delegation + recursion        else:            yield itemgen = (x * x for x in data if x > 0)    # genexpr: no memory for full list

class Money:    __slots__ = ("amount", "currency")          # no __dict__: less RAM, no                                                # dynamic attrs    def __init__(self, amount, currency):        self.amount, self.currency = amount, currency    def __repr__(self):                         # unambiguous, for devs        return f"Money({self.amount!r}, {self.currency!r})"    def __str__(self):                          # readable, for users        return f"{self.amount} {self.currency}"    def __eq__(self, other):        if not isinstance(other, Money):            return NotImplemented               # NOT False — lets other side try        return (self.amount, self.currency) == (other.amount, other.currency)    def __hash__(self):                         # defining __eq__ sets        return hash((self.amount, self.currency))  # __hash__ = None unless you do this    def __add__(self, other): ...    def __lt__(self, other): ...                # + functools.total_ordering

from typing import Protocol, TypeVar, Generic, Literal, TypedDict, overloadfrom collections.abc import Iterable, Callable, Sequencedef find(items: Sequence[str], target: str) -> int | None: ...   # 3.10+ unionsT = TypeVar("T")def first(items: Iterable[T], default: T | None = None) -> T | None: ...class Repository(Protocol):            # structural typing — no inheritance needed    def get(self, id: int) -> str: ...    def save(self, id: int, data: str) -> None: ...def sync(repo: Repository) -> None: ...   # anything with get/save matchesclass Config(TypedDict):    host: str    port: intMode = Literal["r", "w", "a"]class Stack(Generic[T]):               # or `class Stack[T]:` in 3.12+    def push(self, item: T) -> None: ...    def pop(self) -> T: ...

class A: ...class B(A): ...class C(A): ...class D(B, C): ...          # MRO: D → B → C → A → object (C3 linearization)D.__mro__class Base:    def __init_subclass__(cls, **kwargs):       # runs for every subclass —        super().__init_subclass__(**kwargs)     # 90% of metaclass use cases        registry[cls.__name__] = cls

class Positive:                                  # reusable validation descriptor    def __set_name__(self, owner, name): self.name = "_" + name    def __get__(self, obj, objtype=None): return getattr(obj, self.name)    def __set__(self, obj, value):        if value <= 0: raise ValueError(f"{self.name} must be > 0")        setattr(obj, self.name, value)class Order:    price = Positive()    qty = Positive()

from dataclasses import dataclass, field@dataclass(frozen=True, slots=True)              # immutable + memory-efficientclass Point:    x: float    y: float    tags: list[str] = field(default_factory=list)  # NEVER a mutable default

import contextlib@contextlib.contextmanagerdef transaction(conn):    conn.begin()    try:        yield conn          # ← body of the with-block runs here        conn.commit()    except BaseException:        conn.rollback()        raise               # re-raise: don't silently swallowwith contextlib.ExitStack() as stack:            # dynamic number of resources    files = [stack.enter_context(open(p)) for p in paths]    # all closed in reverse order, even on errorwith contextlib.suppress(FileNotFoundError):     # cleaner than try/except/pass    os.remove(path)

class AppError(Exception): ...                   # domain hierarchyclass NotFound(AppError): ...class UpstreamError(AppError): ...try:    resp = call_upstream()except ConnectionError as e:    raise UpstreamError("billing service down") from e   # keeps cause chain    # `from None` hides the chain; bare `raise` re-raises currenttry:    ...except (ValueError, KeyError) as e: ...except Exception:    logger.exception("unexpected")   # logs traceback automatically    raiseelse:    ...                              # ran only if no exceptionfinally:    ...                              # always (even on return)# 3.11+: exception groups (e.g. from TaskGroup)try:    async with asyncio.TaskGroup() as tg: ...except* ValueError as eg:            # except* matches INSIDE the group    for exc in eg.exceptions: ...

import pytestfrom unittest.mock import Mock, AsyncMock, patch@pytest.fixturedef db():    conn = make_test_db()    yield conn                     # teardown after the yield    conn.close()@pytest.mark.parametrize("raw,expected", [("1", 1), ("-3", -3), ("07", 7)])def test_parse(raw, expected):    assert parse(raw) == expecteddef test_invalid():    with pytest.raises(ValueError, match="invalid"):        parse("abc")def test_service(monkeypatch):    monkeypatch.setenv("API_KEY", "test")    monkeypatch.setattr("myapp.client.fetch", lambda url: {"ok": True})@patch("myapp.service.EmailClient")   # patch where it's USED, not where defineddef test_signup(MockEmail):    MockEmail.return_value.send.assert_called_once_with(to="[email protected]")@pytest.mark.asyncio                  # pytest-asyncioasync def test_fetch():    client = AsyncMock()    client.get.return_value = {"ok": True}    assert await fetch_data(client) == {"ok": True}

python -m timeit -s "data=list(range(1000))" "sum(data)"python -m cProfile -s cumulative script.py

def make_counter():    count = 0    def inc():        nonlocal count          # without it: UnboundLocalError        count += 1        return count    return inc# THE classic gotcha — late binding: closures capture variables, not valuesfns = [lambda: i for i in range(3)][f() for f in fns]              # [2, 2, 2] !fns = [lambda i=i: i for i in range(3)]     # default arg captures value → [0, 1, 2]from functools import partiallog_error = partial(log, level="ERROR")     # partial application

myproject/├── pyproject.toml          # single source of truth (PEP 621)├── src/mypkg/              # src layout: forces installed-package imports,│   ├── __init__.py         # prevents "works on my machine" path accidents│   └── core.py└── tests/

[project]name = "mypkg"requires-python = ">=3.11"dependencies = ["httpx>=0.27"][project.scripts]mypkg = "mypkg.cli:main"        # entry point → console command[project.optional-dependencies]dev = ["pytest", "mypy", "ruff"]

import logginglogger = logging.getLogger(__name__)     # per-module logger, NEVER the root                                         # logger in library code# In the APP entry point only (libraries never configure handlers):logging.basicConfig(    level=logging.INFO,    format="%(asctime)s %(name)s %(levelname)s %(message)s",)logger.info("user %s logged in", user_id)   # lazy %-formatting: no f-string                                            # cost when level is filtered outlogger.exception("payment failed")          # inside except: adds traceback

class Model(type):                       # a metaclass is the type of a class    def __new__(mcls, name, bases, ns):        cls = super().__new__(mcls, name, bases, ns)        cls._fields = [k for k in ns if not k.startswith("_")]        return clsclass User(metaclass=Model):    name: strtype(obj) → its class;  type(cls) → its metaclass (usually `type`)Widget = type("Widget", (Base,), {"size": 10})   # classes created at runtime

from sqlalchemy import create_enginefrom sqlalchemy.orm import Session, sessionmakerengine = create_engine(                       # one per process, holds the pool    "postgresql+psycopg://user:pw@localhost/app",    echo=False,          # True logs every SQL statement — best debugging tool here    pool_size=5,    pool_pre_ping=True,  # drop dead connections instead of erroring on first use)SessionLocal = sessionmaker(engine, expire_on_commit=False)with Session(engine) as session:              # closes (and rolls back) on exit    session.add(obj)    session.commit()with Session(engine) as session, session.begin():   # commit on success,    session.add(obj)                                # rollback on exception

from datetime import datetimefrom sqlalchemy import ForeignKey, String, UniqueConstraint, funcfrom sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationshipclass Base(DeclarativeBase):    passclass Author(Base):    __tablename__ = "authors"    id: Mapped[int] = mapped_column(primary_key=True)    name: Mapped[str] = mapped_column(String(100), index=True)    bio: Mapped[str | None]                    # Optional → NULL allowed    created_at: Mapped[datetime] = mapped_column(server_default=func.now())    books: Mapped[list["Book"]] = relationship(        back_populates="author", cascade="all, delete-orphan"    )    __table_args__ = (UniqueConstraint("name", name="uq_authors_name"),)class Book(Base):    __tablename__ = "books"    id: Mapped[int] = mapped_column(primary_key=True)    title: Mapped[str]    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"), index=True)    author: Mapped["Author"] = relationship(back_populates="books")

# one-to-many  →  the "many" side owns the FKclass Author(Base):    books: Mapped[list["Book"]] = relationship(back_populates="author")class Book(Base):    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))    author: Mapped["Author"] = relationship(back_populates="books")# one-to-one  →  same thing with uselist=False on the collection sideprofile: Mapped["Profile"] = relationship(back_populates="author", uselist=False)# many-to-many  →  a plain association tablefrom sqlalchemy import Column, Tablebook_tags = Table(    "book_tags", Base.metadata,    Column("book_id", ForeignKey("books.id", ondelete="CASCADE"), primary_key=True),    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),)class Book(Base):    tags: Mapped[list["Tag"]] = relationship(secondary=book_tags, back_populates="books")

from sqlalchemy import func, selectstmt = select(Book).where(Book.title.ilike("%python%")).order_by(Book.title).limit(10)books = session.scalars(stmt).all()          # list[Book]book  = session.scalars(stmt).first()        # Book | Nonebook  = session.get(Book, 1)                 # by PK — hits the identity map firstsession.scalars(select(Book).where(Book.id == 1)).one()          # raises if != 1 rowsession.scalars(select(Book).where(Book.id == 1)).one_or_none()  # raises if > 1# filtersBook.title == "x"                            # ==, !=, <, >Book.id.in_([1, 2, 3])Book.bio.is_(None)                           # never `== None`Book.title.like("a%"), Book.title.ilike("a%")from sqlalchemy import and_, or_, not_select(Book).where(or_(Book.title == "a", Book.title == "b"))select(Book).where(Book.title == "a", Book.id > 5)     # commas = AND# joins & aggregatesselect(Author).join(Author.books).where(Book.title == "x").distinct()select(Author).outerjoin(Author.books)select(Author.name, func.count(Book.id)).join(Book).group_by(Author.id).having(    func.count(Book.id) > 2)session.scalar(select(func.count()).select_from(Book))        # a number, not a Bookrows = session.execute(select(Book.id, Book.title)).all()     # list[Row] of tuplesfor book_id, title in rows: ...

from sqlalchemy.orm import contains_eager, joinedload, raiseload, selectinload# THE bug: 1 query for the authors, then 1 MORE per author when .books is touchedfor author in session.scalars(select(Author)):    print(author.books)                       # ← N+1# collections → selectinload: 2 queries total, a second SELECT ... WHERE id IN (...)session.scalars(select(Author).options(selectinload(Author.books))).all()# many-to-one / one-to-one → joinedload: single query with a LEFT OUTER JOINsession.scalars(select(Book).options(joinedload(Book.author))).all()# nested, one level deeperselect(Author).options(selectinload(Author.books).selectinload(Book.tags))# joinedload on a COLLECTION duplicates parent rows — .unique() is mandatorysession.scalars(select(Author).options(joinedload(Author.books))).unique().all()# already joining to filter? reuse that join instead of emitting a second onestmt = (    select(Author)    .join(Author.books)    .where(Book.title.ilike("%python%"))    .options(contains_eager(Author.books))    # NB: only the matching books are loaded)# make any accidental lazy load blow up — great in testssession.scalars(select(Author).options(raiseload("*"))).all()

author = Author(name="Ada")session.add(author)session.add_all([book1, book2])session.flush()          # sends INSERT/UPDATE, fills PKs — still inside the transactionsession.commit()         # flush + COMMITsession.rollback()       # discards the transaction AND the in-memory changessession.refresh(author)  # re-SELECT this row nowauthor.name = "Ada L."   # tracked automatically; UPDATE goes out at the next flushsession.delete(author)   # ORM delete: runs cascades and events# set-based, no objects loaded: fast, but skips cascades/events/validatorsfrom sqlalchemy import delete, updatesession.execute(update(Book).where(Book.author_id == 1).values(archived=True))session.execute(delete(Book).where(Book.id.in_(ids)))# upsert (postgres)from sqlalchemy.dialects.postgresql import insertstmt = insert(Tag).values(name="python").on_conflict_do_nothing(index_elements=["name"])session.execute(stmt)

with Session(engine) as session:    with session.begin():                # commit at the end, rollback on exception        session.add(obj)    with session.begin_nested():         # SAVEPOINT — inner failure, outer survives        session.add(risky)with engine.begin() as conn:             # Core, no ORM/session involved    conn.execute(text("UPDATE books SET archived = true"))

alembic init migrations                    # -t async for an async setupalembic revision --autogenerate -m "add books table"alembic upgrade head                       # apply everything pendingalembic downgrade -1                       # step back one revisionalembic current                            # what the DB thinks it is atalembic history --verbosealembic heads                              # more than one = branches to mergealembic merge -m "merge heads" headsalembic stamp head                         # mark as applied WITHOUT running it

# migrations/env.py — autogenerate compares this metadata to the live databasefrom myapp.models import Base                # importing every model matters:target_metadata = Base.metadata             # unimported tables look like drops# a revision filerevision = "a1b2c3d4"down_revision = "9f8e7d6c"def upgrade() -> None:    op.create_table(        "tags",        sa.Column("id", sa.Integer(), primary_key=True),        sa.Column("name", sa.String(50), nullable=False),    )    op.add_column("books", sa.Column("archived", sa.Boolean(), nullable=False,                                     server_default=sa.false()))    op.create_index("ix_books_author_id", "books", ["author_id"])def downgrade() -> None:    op.drop_index("ix_books_author_id", "books")    op.drop_column("books", "archived")    op.drop_table("tags")

pytest                          # everything under testpaths / cwdpytest tests/test_api.py        # one filepytest tests/test_api.py::test_login          # one test  (node id)pytest tests/test_api.py::TestAuth::test_loginpytest "tests/test_api.py::test_login[admin]" # one parametrized casepytest -k "login and not slow" # select by name expressionpytest -m slow                  # select by markerpytest -x                       # stop at the first failurepytest --maxfail=3pytest --lf                     # rerun ONLY last-failedpytest --ff                     # last-failed first, then the restpytest -q                       # quiet   (-v verbose, -vv full diffs)pytest -s                       # don't capture stdout — print() shows uppytest --pdb                    # drop into the debugger on failurepytest --durations=10           # the 10 slowest testspytest --collect-only           # what WOULD run, without running itpytest -n auto                  # parallel  (pytest-xdist)pytest --cov=mypkg --cov-report=term-missing   # coverage (pytest-cov)

import pytestdef test_basics():    assert result == 42                      # plain assert: pytest rewrites it    assert "key" in payload                  # to show both sides on failure    assert user.roles == ["admin", "staff"]def test_raises():    with pytest.raises(ValueError, match="invalid token"):   # match is a REGEX        parse("nope")def test_exception_detail():    with pytest.raises(HTTPError) as exc_info:        client.get("/missing")    assert exc_info.value.status_code == 404 # inspect it AFTER the blockdef test_floats():    assert 0.1 + 0.2 == pytest.approx(0.3)    assert [0.1, 0.2] == pytest.approx([0.1, 0.2], rel=1e-6)def test_warns():    with pytest.warns(DeprecationWarning):        old_api()def test_no_raise():                         # asserting something does NOT raise    parse("ok")                              # just call it; a raise fails the test

import pytest@pytest.fixturedef db():    conn = connect(":memory:")    yield conn                     # everything after yield is teardown    conn.close()                   # runs even if the test fails@pytest.fixture(scope="session")   # created once for the whole rundef engine():    return create_engine(TEST_URL)@pytest.fixture(autouse=True)      # applied to every test in scope, no arg neededdef reset_state():    cache.clear()@pytest.fixturedef user(db):                      # fixtures can request other fixtures    return db.create_user(name="Ada")def test_login(user, db):          # ask for what you need, by name    assert db.authenticate(user) is True# a factory when each test wants its own variant@pytest.fixturedef make_user(db):    def _make(**kwargs):        return db.create_user(**{"name": "Ada", **kwargs})    return _makedef test_admin(make_user):    assert make_user(role="admin").is_admin

def test_files(tmp_path):                 # a fresh Path, per test    config = tmp_path / "config.toml"    config.write_text("debug = true")    assert load(config).debug is Truedef test_env(monkeypatch):    monkeypatch.setenv("API_KEY", "test-key")    monkeypatch.delenv("PROXY", raising=False)    monkeypatch.setattr("myapp.client.fetch", lambda url: {"ok": True})    monkeypatch.chdir(tmp_path)           # all undone automatically at teardowndef test_output(capsys):    greet("Ada")    captured = capsys.readouterr()        # consumes the buffer    assert captured.out == "hello Ada\n"    assert captured.err == ""def test_logs(caplog):    with caplog.at_level(logging.WARNING):        charge(amount=-1)    assert "negative amount" in caplog.text    assert caplog.records[0].levelname == "WARNING"

@pytest.mark.parametrize("raw,expected", [("1", 1), ("-3", -3), ("07", 7)])def test_parse(raw, expected):    assert parse(raw) == expected# readable failure names instead of raw[0], raw[1]...@pytest.mark.parametrize(    "path,status",    [("/", 200), ("/missing", 404), ("/admin", 403)],    ids=["root", "missing", "forbidden"],)def test_routes(client, path, status):    assert client.get(path).status_code == status# mark ONE case without splitting the test@pytest.mark.parametrize("value", [1, 2, pytest.param(0, marks=pytest.mark.xfail)])def test_reciprocal(value):    assert 1 / value# stacking multiplies: 3 x 2 = 6 tests@pytest.mark.parametrize("role", ["admin", "staff", "guest"])@pytest.mark.parametrize("method", ["GET", "POST"])def test_matrix(role, method): ...# a parametrized FIXTURE: every test using it runs once per param@pytest.fixture(params=["sqlite", "postgres"])def db(request):    return connect(request.param)

@pytest.mark.slow                              # custom markerdef test_full_import(): ...@pytest.mark.skip(reason="flaky upstream, see #431")def test_broken(): ...@pytest.mark.skipif(sys.version_info < (3, 12), reason="needs 3.12 batched()")def test_new_syntax(): ...@pytest.mark.xfail(reason="known bug #92", strict=True)def test_known_bug(): ...                      # strict: PASSING is now a failure@pytest.mark.xfail(raises=TimeoutError)        # only THIS error counts as expecteddef test_flaky_network(): ...def test_conditional_skip():    if not shutil.which("psql"):        pytest.skip("psql not installed")      # skip at runtime

[tool.pytest.ini_options]markers = ["slow: takes more than a second", "integration: needs docker"]addopts = "--strict-markers"      # a typo'd marker becomes an error, not a silent skip

from unittest.mock import AsyncMock, MagicMock, Mock, patchdef test_sends_email(monkeypatch):    sent = []    monkeypatch.setattr("myapp.signup.send_email", lambda **kw: sent.append(kw))    signup("[email protected]")    assert sent == [{"to": "[email protected]", "template": "welcome"}]@patch("myapp.signup.EmailClient")             # patch where it is USED,def test_with_mock(MockEmail):                 # not where it is defined    signup("[email protected]")    MockEmail.return_value.send.assert_called_once_with(to="[email protected]")def test_mock_api():    client = Mock(spec=EmailClient)             # spec: typo'd attributes raise    client.send.return_value = {"id": "msg_1"}    client.send.side_effect = TimeoutError      # or: raise on call    assert client.send.call_count == 1    client.send.assert_called_once_with(to="[email protected]")    args, kwargs = client.send.call_argsasync def test_async_client():    client = AsyncMock()    client.get.return_value = {"ok": True}    assert await fetch(client) == {"ok": True}

import pytest@pytest.mark.asyncio                 # pytest-asyncioasync def test_fetch():    assert await fetch_user(1) == {"id": 1}@pytest.fixtureasync def client():                  # async fixtures work the same way    async with AsyncClient() as c:        yield c

[tool.pytest.ini_options]asyncio_mode = "auto"       # every async test is collected, no marker needed

[tool.pytest.ini_options]testpaths = ["tests"]addopts = "-q --strict-markers --strict-config"filterwarnings = ["error"]            # warnings become failures — catches deprecations earlypythonpath = ["src"]

myproject/├── src/mypkg/├── tests/│   ├── conftest.py          # shared fixtures, no import needed│   ├── unit/│   └── integration/│       └── conftest.py      # fixtures for this subtree only└── pyproject.toml