Python quick reference

b = a

Binds a second name to the same object — no copy is made. Use list(a), a[:] or a.copy() for a shallow copy, and copy.deepcopy(a) when the object nests.

b = a; b.append(8) # a changed too

int · float · complex

int is exact and unbounded, float is an IEEE-754 double, complex pairs two floats. / returns a float; // floors toward negative infinity.

-7 // 2 # -4

"".join(parts)

Concatenate an iterable of strings in a single pass. The linear way to build a string, because str is immutable and += rebuilds it every time.

", ".join(["python", "async"]) # 'python, async'

f"{expr!conv:spec}"

Evaluate expr, optionally convert with !r/!s/!a, then format with spec (width, precision, alignment, type).

f"{total:.2f}" # '59.70'

str.encode('utf-8') / bytes.decode('utf-8')

Cross the text/binary boundary in either direction. bytes is immutable; bytearray is the mutable form.

"café".encode("utf-8")

[a, b, c] · list.append(x) · list.sort()

Ordered, mutable sequence. append adds at the end; sort() reorders in place and returns None, sorted() returns a new list.

skus = ["SKU-8841"]; skus.append("SKU-2210")

(a, b, c) · tuple(iterable)

Immutable sequence. Fixed length, indexed by position, hashable when every element is hashable.

city, latitude, longitude = ("Berlin", 52.52, 13.40)

set(iterable)

Build an unordered collection of unique, hashable items. Combine with | & - ^; test with in.

sorted(set(monday_visitors) & set(tuesday_visitors))

d = {key: value}

A mapping from hashable keys to values, kept in insertion order. d[key] raises KeyError when the key is absent; d.get(key, default) does not.

settings.get("retry_budget", 3)

seq[start:stop]

New sequence from start up to but not including stop. Omitted bounds default to 0 and len(seq); out-of-range bounds clamp instead of raising.

records[1:4] # positions 1, 2, 3

seq[start:stop:step]

Take every step-th item from start toward stop. A negative step walks backwards and swaps which end the omitted bounds mean.

readings[::-1]

host, port = endpoint

Bind several names in one statement by taking an iterable apart position by position. The number of names must equal the number of values, or ValueError.

for resource, amount in limits.items():

first, *rest = items · *leading, last = items · first, *middle, last = items

One starred target per assignment absorbs the leftover items as a list, even when there are none. Plain names are filled first from both ends.

first, *middle, last = line_items

[expression for item in iterable if condition]

One expression that walks an iterable once and builds a container. The brackets choose the container; the loop name stays inside them.

[item["sku"] for item in line_items if item["paid"]]

[expr for item in iterable if cond]

Build a new list in one eager pass. The filter if goes last; a conditional expression goes in the output slot at the front.

[amount for amount in line_items if amount > 0]

{expr for item in iterable}

Build a set from an iterable, dropping duplicates as it goes. Every element produced must be hashable, and the result has no order.

{record["email"].split("@")[1] for record in signups}

{key_expr: value_expr for item in iterable}

Builds a new dict in one expression. A trailing if filters pairs before they land; a repeated key is silently overwritten by the later pair.

{post_id: slug for slug, post_id in slug_to_id.items()}

(expr for item in iterable)

Produces items on demand instead of building a container. Flat memory, one-shot, and no len(), indexing or slicing.

sum(unit_price * quantity for _, unit_price, quantity in line_items)

value_if_true if condition else value_if_false

An expression evaluating to one of two values. else is required, only the chosen branch is evaluated, and it binds looser than arithmetic.

status = "retry" if attempts < retry_budget else "give up"

bool(value) · if value:

Every object has a truth value. False, None, the zeros and every empty container are falsy; everything else is truthy. bool() tries __bool__, then __len__, then answers truthy.

display_name = raw_name or "anonymous"

value is None

None is the sole instance of NoneType — the object meaning "no value". Test for it by identity, and use it as the sentinel default in place of a mutable one.

def add_tag(tag, tags=None): ...

a is b · a == b

is compares identity (same object); == compares value via __eq__. Equal values are not always one object — reserve is for None, True, False and true identity checks.

if cached_row is None: ...

hash(obj)

Returns an int used to file obj in a dict or set. Requires equal objects to return equal hashes, which is why every mutable built-in is unhashable — and why defining __eq__ on a class disables its hash until you supply one that agrees.

cache[user_id, tuple(filters)] = result

mutable → edits in place · immutable → replaces

A mutable object (list, dict, set) can change without becoming a new object, so every alias sees the edit. An immutable one (int, str, tuple) never changes — every "update" rebinds a name to a new object instead.

tags += ["new"] # mutates if tags is a list, rebinds if tags is a tuple

def name(params):

Builds a function object from the indented block and binds name to it. The body does not run, and its names are not checked, until the function is called.

def greet(name, greeting="Hello"): return f"{greeting}, {name}!"

func(value1, value2, ...)

Arguments matched to parameters by position, in order. A missing or extra argument raises TypeError naming the mismatch; a same-typed swap raises nothing — it just runs with the wrong values in the wrong places.

create_user("nova", "nova@example.com", "admin")

func(name1=value1, name2=value2, ...)

Arguments matched to parameters by name, in any order, after any positional arguments. An unrecognized name or one that duplicates a positional fill raises TypeError naming the problem keyword.

create_user(role="admin", username="nova", email="nova@example.com")

def func(param=default):

Makes param optional; omitted calls use default, computed once at def time and reused by every such call. Never use a mutable literal as a default — default to None and build the value inside the body instead.

def add_tag(tag, tags=None): if tags is None: tags = []

def func(pos, *, kw_only1, kw_only2=default):

Everything after a bare * in the parameter list can only be passed by name. Passing one positionally, or omitting one with no default, is a TypeError.

def resize(image, *, width, height): ... resize("photo.jpg", width=800, height=600)

def name(pos_only1, pos_only2, /, normal, *, kw_only):

A bare / marks every parameter before it as reachable only by position — never by name. Lets a function change a parameter name later without breaking any caller, and matches how several builtins (len, pow) already behave.

def power(base, exp, /): return base ** exp

def name(*args):

Collects every extra positional argument into a tuple named args (any name works). The same * unpacks a sequence into separate positional arguments at a call site — two directions, one symbol.

def total(*amounts): return sum(amounts)

def name(**kwargs):

Collects every extra keyword argument into a dict named kwargs (any name works), keyed by the name each was passed with. The same ** unpacks a dict into separate keyword arguments at a call site.

def build_query(**filters): return filters

name = some_function

A function is an ordinary object — assignable, storable in a container, passable as an argument, returnable from another function. A bare name refers to the function object; name() calls it.

loud = shout funcs = [shout, str.lower]

map(f, iterable) · filter(f, iterable) · sorted(items, key=f)

A function that takes a function as an argument, returns one, or both. Built on functions being first-class — a function argument or return value works exactly like any other.

list(filter(lambda n: n > 0, [1, -2, 3, -4, 5]))

def outer(): x = ... def inner(): nonlocal x ... return inner

A nested function that keeps access to variables from its enclosing function even after that function has returned. nonlocal is required to reassign (not just read) a captured variable. Each call to the outer function creates an independent closure.

counter_a = make_counter() counter_a() # 1 counter_a() # 2

lambda args: expression

A nameless, single-expression function. The expression's value returns automatically — no return keyword, no statements, no multiple lines. Most useful passed directly as a short callback (sorted's key=, for example) rather than assigned to a name.

sorted(people, key=lambda p: p[1])

def name(param: Type = default) -> ReturnType:

Optional expressions attached to parameters and the return value, stored on func.__annotations__. Read by type checkers and IDEs — never enforced by the interpreter at call time.

def add(a: int, b: int) -> int: return a + b

func.__name__ · func.__doc__ · func.__dict__

A function built with def is an instance of type function, carrying built-in introspection attributes (__name__, __doc__, __defaults__, __code__) plus its own __dict__ — arbitrary custom attributes can be set on it directly.

greet.call_count = 0 greet.call_count += 1

Local → Enclosing → Global → Built-in

The order Python searches when a name is looked up inside a function — the first scope that defines the name wins. Assigning to a name anywhere in a function makes it local for the function's entire body, which is what causes UnboundLocalError when a name is read before its local assignment runs.

x = "global" def f(): x = "local" # shadows the global x, inside f only

global name

Declares that name, for the rest of the function, refers to the module-level variable — assignments update it directly instead of creating a local. Only needed for reassignment; reading a global works without it.

def increment(): global count count += 1

nonlocal name

Declares that name refers to the nearest enclosing function's variable, not a new local — required to reassign a captured variable inside a closure. Reaches exactly one enclosing function scope, never the module (that is global's job).

def increment(): nonlocal count count += 1

enumerate(iterable, start=0)

Wraps an iterable in a lazy sequence of (index, item) tuples, counting from start. Unpack in a for loop instead of tracking a counter by hand.

for i, item in enumerate(items, start=1): print(i, item)

zip(*iterables)

Pairs items from two or more iterables position by position, lazily, stopping at the shortest one. dict(zip(keys, values)) builds a dict from two parallel lists.

list(zip([1, 2, 3], ["a", "b", "c"]))

sorted(iterable, key=None, reverse=False)

Returns a new sorted list from any iterable, leaving the original unchanged. key names a function to sort by; reverse=True sorts descending. Stable: ties keep their original order.

sorted(words, key=len, reverse=True)

reversed(sequence)

Lazy iterator walking a sequence back to front. Needs a real sequence (supports len() and indexing), leaves the original untouched, unlike the in-place list.reverse().

for item in reversed(items): print(item)

any(iterable) · all(iterable)

any() is True if at least one item is truthy; all() is True only if every item is. Both short-circuit. any([]) is False, all([]) is True.

any(n > 10 for n in nums)

min(iterable, key=None, default=...) · max(iterable, key=None, default=...)

Return the smallest/largest item — or of several separate arguments. key compares a derived value instead of the item itself. Raises ValueError on an empty iterable unless default is given.

max(words, key=len)

len(container) · sum(iterable, start=0)

len() returns item count for any sequence or collection, O(1). sum() adds a numeric iterable to start, left to right — raises TypeError on strings; use str.join instead.

sum(nums, 100)

map(func, iterable, ...)

Applies func to every item lazily, yielding results one at a time. With several iterables, pairs items positionally and stops at the shortest — wrap in list() to materialize.

list(map(str.upper, ["a", "b"]))

filter(func, iterable) · filter(None, iterable)

Keeps items where func(item) is truthy, lazily. filter(None, iterable) drops falsy items directly, with no function — including 0, "", and None alike.

list(filter(lambda n: n > 0, nums))

iter(iterable) · next(iterator, default=...)

iter() returns a fresh iterator from an iterable. next() pulls the next value and advances it, raising StopIteration when exhausted unless a default is given. A for loop is this pair, automated.

it = iter([1, 2]); next(it)

range(stop) · range(start, stop) · range(start, stop, step)

A lazy, immutable sequence of evenly-spaced integers, computed on demand from start/stop/step. stop is always excluded. Supports len(), indexing, and "in" like a real sequence.

list(range(0, 10, 2))

isinstance(obj, type_or_tuple) · issubclass(cls, type_or_tuple)

isinstance checks an object against a type (or tuple of types), including subclasses. issubclass checks a class against a type the same way. bool is a subclass of int.

isinstance(value, (int, float))

getattr(obj, name, default=...) · setattr(obj, name, value) · hasattr(obj, name)

Read/write/check an attribute by name, as a string, rather than dotted syntax — the only way when the name is chosen at runtime. getattr raises AttributeError unless a default is given.

getattr(config, field, "unset")

callable(obj)

True if obj() would be legal — functions, classes, and instances whose type defines __call__. Never calls obj, and says nothing about whether the call would succeed.

callable(list) # True — calling it constructs a new list

import name · import name as alias · from name import thing

A module is any .py file. import runs it once and binds a namespace; from ... import binds one name directly.

from mathy import area as circle_area

package/__init__.py · package.submodule

A directory becomes an importable package once it contains __init__.py. Submodules follow the folder structure, dot by dot.

from shop.utils.formatting import as_currency

package/__init__.py

Marks a directory as a package; runs once on first import. Re-export a submodule name here to expose it directly through the package.

from .pricing import apply_discount # in shop/__init__.py

from pkg.mod import x · from .mod import x · from ..pkg import x

Absolute imports spell the full path; relative imports count leading dots from the current module — one dot for this package, two for its parent.

from .utils.formatting import as_currency

sys.path (search order) · sys.modules (import cache)

import searches sys.path in order and stops at the first match; the result is cached in sys.modules, so re-importing a name never re-runs the file.

'counter' in sys.modules

sys.path · sys.path.append(dir) · sys.path.insert(0, dir)

The ordered list import searches. Starts with the running script's directory; append/insert extend it at runtime.

sys.path.append("/extra/modules")

if __name__ == "__main__":

__name__ is "__main__" when a file runs directly, or the module's own name when imported. Guard script-only logic with this check.

if __name__ == "__main__": main()

import x (survives) · from x import name (fails, usually)

A imports B, B imports A back. from-imports fail on the not-yet-defined name; plain module imports usually survive because the attribute lookup happens later.

import a def func_b(): a.func_a()

class Name: ... · obj = Name()

A class is a blueprint; calling it builds one independent object (instance).

class Dog: pass rex = Dog()

def __init__(self, arg1, arg2): ...

The constructor's setup step — called automatically right after Name(...) builds the object, to set starting attributes on self.

class Dog: def __init__(self, name): self.name = name

self.attr = value

Data that belongs to one instance, stored in that instance's own __dict__ — independent from every other instance of the same class.

rex.age = 3 print(vars(rex))

class Name: attr = value

Data attached to the class itself, shared by every instance that has not set its own attribute of the same name.

class Dog: count = 0 Dog.count += 1

def method(self, arg): ... → obj.method(arg)

A function defined in a class body that receives the instance it was called through as self, automatically.

class Dog: def bark(self): return f"{self.name} says woof"

@classmethod def name(cls, ...): ...

A method bound to the class (cls), not an instance — most often used to build an alternative constructor that also works for subclasses.

@classmethod def from_string(cls, text): return cls(*text.split("-"))

@staticmethod def name(args): ...

A method with no self or cls — an ordinary function kept in the class namespace because it is logically related to the class.

@staticmethod def is_valid_name(name): return bool(name)

self._internal = x · self.__hidden = x

Bundling data with the methods that use it; underscore prefixes signal internal-use-only by convention, not by enforced access control.

class Account: def __init__(self, balance): self.__balance = balance

obj.method() — interface, not implementation

Depending on WHAT a method does rather than HOW it does it, so the internal implementation is free to change without breaking callers.

stack.push(x) # hides whatever storage is used underneath

class Sub(Base): def method(self): return super().method()

A subclass gains every Base attribute and method, can override any of them, and reaches the parent version from inside an override with super().

class Puppy(Dog): def __init__(self, name, weeks_old): super().__init__(name) self.weeks_old = weeks_old

for obj in objs: obj.method()

Calling the same method name on objects of different types and getting each type's own behaviour — no shared base class required (duck typing).

for shape in [Circle(2), Rectangle(3, 4)]: print(shape.area())

self.part = Part() # created and owned here

A "has-a" relationship where the whole creates and owns the part — the part's lifetime is tied entirely to the owner's.

class Car: def __init__(self): self.engine = Engine()

self.part = part # created elsewhere, referenced here

A "has-a" relationship where the whole references a part it did not create — the part can be shared and outlives the whole.

team = Team("Falcons", [existing_player])

self.related = [] # neither owns the other

The general "knows-about/uses" relationship between two independent objects — no ownership in either direction; aggregation and composition are stricter special cases.

student.courses.append(course) course.students.append(student)

is-a → inherit · has-a / varies independently → compose

Inheritance models a fixed "is-a" hierarchy; composition assembles behaviour from swappable parts. Favour composition once more than one independent dimension of variation exists.

class Duck: def __init__(self, fly_behavior): self.fly_behavior = fly_behavior

from abc import ABC, abstractmethod

ABC is the base class to inherit from; @abstractmethod marks a method every concrete subclass must override before it can be instantiated.

class Shape(ABC): @abstractmethod def area(self): ...

class Name(ABC): ...

Inheriting from ABC enables @abstractmethod enforcement — the class (and any subclass missing a required method) cannot be instantiated.

class Shape(ABC): @abstractmethod def area(self): ...

@abstractmethod def name(self): ...

Marks a method as required for every concrete subclass — enforced only inside a class using ABCMeta (via ABC). Stacks innermost with @property/@classmethod/@staticmethod.

class Shape(ABC): @abstractmethod def area(self): ...

obj.method() # no type check, just calls it

Python resolves a method call by whether the object has it, not by its declared class — "if it walks like a duck and quacks like a duck."

def load(source): return source.read() # any object with .read() works

class Name(Protocol): def method(self) -> T: ...

Names a structural type — any class with matching methods satisfies it, no inheritance required. Add @runtime_checkable for isinstance() support.

@runtime_checkable class Readable(Protocol): def read(self) -> str: ...

compatible ⇔ same shape (structural) vs. compatible ⇔ declared relationship (nominal)

Structural typing judges compatibility by an object's methods/attributes; nominal typing requires an explicit inherits-from/implements declaration. Python defaults to structural.

def send_up(x: FlyerProtocol) -> str: return x.fly() # any matching shape works

@property def name(self): return ...

Turns a method into an attribute-style read — obj.name, no parentheses. Runs fresh on every read; read-only unless a setter is added.

@property def area(self): return 3.14159 * self.radius ** 2

@name.setter def name(self, value): self._name = value

Makes a property writable — obj.name = value runs this method, most often to validate or transform the value before storing it.

@celsius.setter def celsius(self, value): if value < -273.15: raise ValueError("below absolute zero") self._celsius = value

@name.deleter def name(self): ...

Defines what del obj.name does — reset the underlying value, clean up, or raise to forbid deletion. Without one, del raises AttributeError.

@total.deleter def total(self): self._total = None # reset, forcing a recompute later

@property def derived(self): return f(self.a, self.b)

A property that computes its value from other attributes on every read, instead of storing its own — it can never go stale, at the cost of recomputing every time.

@property def full_name(self): return f"{self.first_name} {self.last_name}"

def __new__(cls, *args, **kwargs): return super().__new__(cls)

Creates the object, before __init__ sets it up. Rarely overridden — mainly for singletons and subclassing an immutable built-in type.

class Config: _instance = None def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance

def __repr__(self): return f"Name({self.value!r})" def __str__(self): return f"{self.value}"

__repr__ is the unambiguous, developer-facing form (REPL, debugging, containers); __str__ is the reader-facing form used by print()/str(). Missing __str__ falls back to __repr__.

class Point: def __repr__(self): return f"Point({self.x!r}, {self.y!r})"

def __eq__(self, other): if not isinstance(other, Name): return NotImplemented return self.value == other.value

Defines what == compares — by value instead of Python's default identity check. __ne__ is derived automatically. Defining __eq__ disables hashing unless __hash__ is also defined.

p1 = Point(1, 2) p2 = Point(1, 2) p1 == p2 # True, via __eq__

def __lt__(self, other): return self.value < other.value

Defines < (and __le__ defines <=) — neither is derived from the other. @functools.total_ordering fills in all four ordering operators from __eq__ plus just one.

@total_ordering class Money: def __eq__(self, other): ... def __lt__(self, other): ...

def __gt__(self, other): return self.value > other.value

Defines > (and __ge__ defines >=). Neither is derived from __lt__/__le__. Python tries the reflected operator on the other operand before raising TypeError.

a, b = Money(700), Money(200) a > b # True, via __gt__

def __hash__(self): return hash((self.a, self.b))

Returns an int used to place an object in a dict/set. Must be consistent with __eq__: equal objects need equal hashes. Defining __eq__ alone disables hashing.

def __hash__(self): return hash((self.x, self.y)) # same fields as __eq__

def __bool__(self): return self.some_condition

Defines truthiness — bool(obj) and if obj:. Without it, Python falls back to __len__ == 0 being falsy, then to always-truthy if neither is defined.

def __bool__(self): return len(self.items) > 0

def __len__(self): return len(self._items) def __contains__(self, item): return item in self._items

__len__ defines len(obj); __contains__ defines "item in obj". Without __contains__, "in" falls back to a slower linear scan via __iter__.

len(cart) # via __len__ "pen" in cart # via __contains__

def __iter__(self): return self def __next__(self): if done: raise StopIteration return value

__iter__ makes an object iterable; __next__ makes it an iterator. A for loop is exactly iter() once, then next() until StopIteration.

for n in Countdown(3): print(n) # calls __iter__ once, __next__ repeatedly

def __getitem__(self, key): return self._data[key] def __setitem__(self, key, value): self._data[key] = value

Defines obj[key] reads and writes. Independent hooks — a class can support only reads. Raise IndexError/KeyError for an invalid key, by convention.

s["theme"] = "dark" # via __setitem__ s["theme"] # via __getitem__

def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): return False

__enter__ runs at the top of a with block; __exit__ always runs at the end, even on an exception. Returning truthy from __exit__ suppresses it.

with Resource() as r: ... # __enter__ then, always, __exit__

def __call__(self, *args, **kwargs): return self.do_something(*args, **kwargs)

Lets an instance be called like a function — obj(args) is sugar for obj.__call__(args). Useful when state needs to persist across calls.

double = Multiplier(2) double(5) # 10, via __call__

class C(A, B): def method(self): super().method() # next in C.__mro__

Multiple inheritance gives a class several bases; the MRO (C3 linearization) is the single deterministic lookup order across them. super() follows that order, which is what makes cooperative inheritance work.

print(Diamond.__mro__) # (Diamond, Left, Right, Base, object)

class Widget(SomeMixin, Base): pass

A mixin is a small class adding one focused behaviour, meant to be combined via multiple inheritance rather than instantiated alone. List it before the real base class.

class LoggedWidget(LoggingMixin, Widget): pass

class Descriptor: def __get__(self, obj, owner): ... def __set__(self, obj, value): ...

A class implementing __get__/__set__, assigned as a CLASS attribute on another class, to intercept attribute access. @property is built from this protocol.

class Product: price = PositiveNumber() # reusable validation descriptor

def __getattr__(self, name): raise AttributeError(name) # fallback only def __setattr__(self, name, value): super().__setattr__(name, value)

__getattribute__ intercepts every read; __getattr__ only fires when normal lookup already failed; __setattr__ intercepts every write. Delegate to super() to avoid infinite recursion.

obj.x = 1 # __setattr__ runs obj.missing # __getattr__ runs, only if missing

class Name: __slots__ = ("a", "b")

Declares the only attribute names a class instance may hold, removing the per-instance __dict__ — lower memory, faster access, no arbitrary new attributes.

class Point: __slots__ = ("x", "y")

class Meta(type): def __new__(mcls, name, bases, namespace): return super().__new__(mcls, name, bases, namespace)

A metaclass is the class of a class — type builds every ordinary class by default. metaclass=Meta lets Meta control how a class itself is constructed.

class Name(metaclass=Meta): pass

type(x) · isinstance(x, object)

Every Python value is an object — a number, string, function, module, or class. type(x) reports the class that built it; isinstance(x, object) is always True.

type(5) # <class 'int'>

x = obj · y = x · del x

A name references an object rather than containing it. Assignment binds a name to an existing object; del removes a name, and the object survives if any other name still references it.

y = x del x print(y) # still works — the object outlived the name x

id(obj)

Returns an int unique among currently-alive objects, constant for the object's lifetime. In CPython, it is the object's memory address — but that is an implementation detail, and a freed id can be reused by a later object.

a is b # shorthand for id(a) == id(b)

def __eq__(self, other): ...

a == b calls a.__eq__(b). The default, inherited from object, falls back to identity — a class must override __eq__ before instances with equal fields compare equal.

def __eq__(self, other): if not isinstance(other, Point): return NotImplemented return self.x == other.x and self.y == other.y

immutable: int, str, tuple, frozenset · mutable: list, dict, set

Mutability is decided by an object's type. An immutable object never changes after creation; a mutable one can. An immutable container (tuple) can still hold a mutable item that changes.

t = (1, [2, 3]) t[1].append(4) # legal — the list changes, the tuple does not

hash(obj) · object.__hash__ (id-based default)

A plain class's inherited __hash__ is based on id(), not its attributes — two equal-looking instances hash differently. Defining __eq__ sets __hash__ to None until a matching one is supplied.

class Tag: def __eq__(self, other): ... def __hash__(self): return hash(self.name) # must agree with __eq__

__new__ -> __init__ -> (referenced) -> __del__

An object is created by __new__/__init__, stays alive while its reference count is above zero, and is destroyed the instant that count reaches zero — running __del__ first, if the class defines one.

obj = Tracked("x") # created del obj # refcount 0 -> __del__ runs

sys.getrefcount(obj)

Reports how many references point at obj, including a temporary +1 for getrefcount's own argument. An object is freed the instant its true count reaches zero — CPython's primary memory-management mechanism.

import sys sys.getrefcount(obj) - 1 # the real count, before this call

import gc · gc.collect() · gc.get_threshold()

The gc module runs a separate, periodic collector that finds and frees reference cycles — objects that only reference each other, which reference counting alone can never bring to zero.

gc.collect() # run now; returns the number of objects freed

a.other = b; b.other = a

A reference cycle: objects referencing each other in a loop. Reference counting alone cannot free one — either the gc module's cycle collector runs eventually, or a weakref in one direction avoids creating the cycle at all.

child._parent = weakref.ref(parent) # avoids the cycle up front

sys.getsizeof(obj)

Reports the bytes obj occupies in CPython's private heap, including fixed per-object overhead (refcount, type pointer). Small objects (≤ 512 bytes) are served by pymalloc's arena allocator rather than a per-object malloc() call.

sys.getsizeof([]) # 56 — overhead alone, holding nothing

stack: frames + references · heap: every object

Python's call stack holds frames — one per active call — each containing references into the heap, where every object actually lives. Unlike C, there is no stack-allocated object data to reason about; a returned reference is always safe.

def f(): x = [1, 2, 3] # x: a frame reference; the list: on the heap return x # safe — the list never depended on the frame

weakref.ref(obj) · weakref.WeakValueDictionary()

Creates a reference that does not count toward obj's refcount. Call it to get the live object, or None once obj has been freed. WeakValueDictionary auto-removes entries whose value has been freed.

r = weakref.ref(obj) r() # obj, or None if freed

sys.intern(s)

Registers a string in CPython's shared intern pool, guaranteeing that future equal strings passed through sys.intern() share one object. Small ints (-5..256) and some string literals are interned automatically — never rely on that alone for correctness.

sys.intern("a") is sys.intern("a") # True, guaranteed

copy.copy(x) · copy.deepcopy(x)

copy.copy() builds a new outer container but shares nested objects with the original. copy.deepcopy() recursively copies every level, so nothing is shared. Plain assignment (y = x) copies nothing at all — it is a second name for the same object.

shallow["items"] is original["items"] # True — shared deep["items"] is original["items"] # False — independent

f(args) -> new frame -> bind params -> run body -> return

Calling a function creates a new frame, binds each argument to its parameter name inside it, runs the body using that frame's local namespace, then discards the frame and returns the result.

def add(a, b): return a + b add(2, 3) # 5 — frame created, used, discarded

sys._getframe() · frame.f_back · frame.f_locals

A frame holds one call's local variables, current line, and a link (f_back) to its caller's frame. The call stack is the chain of active frames, growing per call and shrinking per return — a traceback is that chain printed out.

frame = sys._getframe() frame.f_back.f_code.co_name # the caller's function name

globals() · locals()

globals() is the real, live module-level namespace — safe to mutate. locals(), inside a function, is a snapshot dict rebuilt on each call — mutating it does not reliably affect real local variables.

globals()["x"] = 1 # a real edit locals()["x"] = 1 # inside a function: usually has no effect

func.__closure__ · cell.cell_contents

func.__closure__ is a tuple of cell objects, one per captured variable, matching func.__code__.co_freevars. Functions from the same enclosing call share a cell; separate calls each get their own.

add5.__closure__[0].cell_contents # 5 — the captured value of n

func.__code__.co_freevars

Lists the names a function reads (never assigns) that come from an enclosing function's scope — decided at compile time by scanning the whole function body. An assignment anywhere in the body makes a name local instead.

inner.__code__.co_freevars # ('value',) — value is read but never assigned in inner

lambda x=x: ... # freezes late-bound x at definition time

A function body's names are looked up when the function runs, not when it is defined — every call re-resolves them fresh. This is why closures over a loop variable all see its final value, unless frozen via a default argument.

handlers.append(lambda event_type=event_type: ...) # captures THIS iteration's value

import dis · dis.dis(func)

Python source compiles to bytecode before running, executed by CPython's stack-based virtual machine. dis.dis() shows a function's bytecode. Not guaranteed stable across Python versions or implementations — CPython is the reference implementation, not the only one.

dis.dis(add) # prints the compiled bytecode instructions for add

def f(x): return g(x) # no globals, no mutation, no I/O

A pure function returns a value that depends only on its arguments and has no side effects — no global write, no argument mutation, no I/O. Same input always gives the same output.

def add_tax(price, rate): return price * (1 + rate)

compose(f, g) = lambda x: f(g(x))

Chains functions so each one's output feeds the next one's input. Python has no built-in compose() — write a small helper, or fold a list of functions with reduce().

shout = compose(str.upper, str.strip) shout(" hi ") # 'HI'

functools.reduce(func, iterable, initial=...)

Folds an iterable into one value by calling func(accumulator, item) left to right. Without initial, the first item seeds the accumulator; an empty iterable with no initial raises TypeError. Always eager, never lazy.

reduce(lambda acc, n: acc + n, [1, 2, 3, 4], 0) # 10

import functools

The standard-library module for function-level tools: reduce (folding), partial (argument binding), lru_cache/cache (memoization), wraps (preserving metadata through a decorator), and total_ordering.

from functools import reduce, partial, lru_cache

functools.partial(func, *args, **kwargs)

Returns a new callable with some of func's arguments already bound. Positional arguments fill the earliest unbound slots; keyword arguments bind by name and can still be overridden at call time unless func forbids it.

square = partial(power, exponent=2) square(5) # 25

map(...) · filter(...) · (x for x in ...) · a generator function

A lazy expression is built instantly and computes nothing until iterated. A list comprehension is the opposite — eager, computing every item immediately.

gen = (n * n for n in big_source) # instant; next(gen) computes one item

stage_two(stage_one(source))

Chains generator functions so each stage yields to the next, one item at a time. No intermediate list is ever built — memory stays flat regardless of source size, as long as every stage stays lazy.

pipeline = positive_only(parse(nonempty(strip_lines(read_lines()))))

tuple · frozenset · namedtuple · @dataclass(frozen=True)

Data structures that refuse mutation after creation, so a value can be shared without defensive copying. Nesting a mutable type (a list) inside one of these does not make that inner object immutable too.

@dataclasses.dataclass(frozen=True) class Config: host: str

@my_decorator def process(): ...

Sugar for process = my_decorator(process), run once at def time. my_decorator returns a wrapper that stands in for the original function.

def my_decorator(func): def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper

def outer(config): def decorator(func): def wrapper(*args, **kwargs): ... return wrapper return decorator

A decorator that takes its own arguments needs three nested functions — the outer one for the decorator's config, then decorator(func), then wrapper(...).

@repeat(times=3) def greet(name): return f"Hi, {name}"

def factory(config): def decorator(func): ... return wrapper return decorator

A plain function that returns a decorator. Calling it with different arguments produces independent decorators, each closing over its own configuration.

@validate_range(0, 100) def set_volume(level): return f"volume set to {level}"

class Name: @my_decorator def method(self, ...): ...

A decorator applies to a method exactly like a plain function — self simply arrives as the first positional argument inside wrapper.

@log_call def deposit(self, amount): self.balance += amount

@my_decorator class Name: ...

Sugar for Name = my_decorator(Name) — the decorator receives the class object itself and must return a class, usually the same one modified.

def add_repr(cls): cls.__repr__ = ... return cls

@bold @italic def f(): ... # == f = bold(italic(f))

Multiple decorators apply bottom-up at decoration time (closest to def first) but run outer-to-inner at call time (topmost decorator's wrapper runs first).

@bold @italic def shout(text): return text.upper()

@functools.wraps(func) def wrapper(*args, **kwargs): ...

Copies __name__, __doc__ and __module__ from func onto wrapper, and sets wrapper.__wrapped__ = func. Without it, a decorated function reports wrapper's identity instead of its own.

import functools @functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs)

class Name: def __init__(self, func): self.func = func def __call__(self, *args, **kwargs): return self.func(*args, **kwargs)

A class implementing __call__ can act as a decorator — @Name above a def runs Name(func), and the resulting instance replaces the function, with state stored as plain attributes on self.

@CountCalls def say_hello(name): return f"Hello, {name}"

iterable: __iter__() -> iterator · iterator: __next__() -> value

An iterable produces iterators; an iterator produces values. iter(iterable) gets a fresh iterator; next(iterator) advances it.

it = iter([1, 2, 3]); hasattr(it, "__next__") # True

__iter__(self) -> iterator · __next__(self) -> value | raise StopIteration

The two-method contract behind for, list(), sum(), and unpacking. A generator function satisfies it automatically.

list(count_up_gen(3)) # works exactly like list(CountUpIterator(3))

def name(): yield value

Any function containing yield becomes a generator function. Calling it builds a paused generator; next() runs it to the next yield.

gen = read_batches(7, 3); next(gen) # [0, 1, 2]

yield from iterable

Delegates to iterable, yielding each of its items and forwarding send()/throw()/close() to it.

def flatten(x): for i in x: yield from flatten(i) if isinstance(i, list) else [i]

[expr for x in data] (eager) · (expr for x in data) (lazy)

Square brackets compute every item immediately; round brackets compute each item only when asked.

sum(transform(x) for x in data) # transform() runs one x at a time, inside sum()

stage_n(...stage_2(stage_1(source)))

Nested generator calls compose into a pipeline: each item flows through every stage before the next item enters.

list(filter_even(parse_ints(read_lines(raw_lines))))

sys.getsizeof(a_generator) # constant, regardless of item count

A generator (or a pipeline of them) holds one item at a time — flat memory. A list holds every item at once — memory scales with count.

sum(1 for _ in (f(x) for x in huge_source)) # never builds a list

gen.send(value)

Resumes a paused generator, delivering value as the result of its current yield expression. Requires priming with next() first.

next(gen); gen.send(10) # gen must already be paused on a yield

gen.close()

Raises GeneratorExit at the generator's current pause point, running any finally block, then leaves it exhausted.

gen = managed_resource(); next(gen); gen.close() # finally block runs

BaseException > Exception > ValueError, KeyError, OSError, ...

Every exception inherits BaseException; Exception is the subclass for ordinary, handleable errors. SystemExit/KeyboardInterrupt/GeneratorExit sit outside Exception.

issubclass(KeyboardInterrupt, Exception) # False

try: ... except SomeError as e: ...

try wraps code that might raise; except names the exception type to catch. Name a specific type rather than a bare except:.

try: result = 10 / n except ZeroDivisionError: result = None

try: ... except E: ... else: ... # only if try raised nothing

else runs only when the try block completed without raising — keeps success-path code from being caught by except.

try: value = int(text) except ValueError: value = None else: print(value * 2)

try: ... finally: cleanup()

finally always runs, regardless of whether the try block succeeded, raised, or returned — the place for guaranteed cleanup.

f = open(path) try: process(f) finally: f.close()

except (TypeA, TypeB) as e: ...

Catches either exception type with one shared handler. Requires parentheses — a bare comma is Python 2 syntax and raises SyntaxError in Python 3.

except (TypeError, ValueError) as e: return f"bad input: {e}"

raise SomeError("message")

Constructs and raises an exception in one step; a bare raise inside except re-raises the current exception unchanged.

if amount < 0: raise ValueError(f"amount cannot be negative: {amount}")

raise New(msg) from original

Explicitly sets New.__cause__ = original; the traceback shows both, labelled "the direct cause of the following exception."

raise ConfigError("invalid config value") from e

except A: raise B(...) # __context__ set automatically, no `from` needed

Any raise inside an active except block sets __context__, even with no explicit `from` — the traceback reads "During handling of the above exception."

raise NewError("cleanup also failed") from None # suppress the implicit chain

class MyError(Exception): def __init__(self, ...): super().__init__(message)

Subclass Exception to define a project-specific error type; override __init__ to attach structured data and still call super().__init__(message).

raise InsufficientFundsError(balance=100, amount=150)

def boundary(...): try: return inner_call(...) except SpecificError as e: raise StableError(...) from e

The one place in a call chain — an entry point — that catches internal exceptions broadly and translates them into a stable, intentional response.

raise ExternalAPIError(f"upstream call failed: {e}") from e

# no try/except needed — this is the default behaviour

An exception with no matching except in the current frame automatically continues up the call stack to the nearest caller that has one.

def parse_row(row): return int(row) # lets ValueError propagate on purpose

RETRYABLE = (ConnectionError, TimeoutError, RateLimitError)

Errors caused by a transient condition, where retrying the identical call has a real chance of succeeding. Defined once as a fixed set, reused everywhere.

if isinstance(exc, RETRYABLE): retry()

NON_RETRYABLE = (ValueError, InvalidRequestError)

Errors caused by the request itself, not a transient condition — retrying the identical call cannot succeed. Fail fast rather than looping.

if isinstance(exc, NON_RETRYABLE): raise # surface immediately, do not retry

requests.get(url, timeout=5)

Bounds how long an operation may run before raising a timeout error — without one, a hung call can block indefinitely.

try: requests.get(url, timeout=5) except requests.Timeout: handle_timeout()

for attempt in range(1, attempts + 1): try: return fn() except RETRYABLE: ...

A retry loop needs a hard cap on attempts, a delay between them, and a re-raise of the final failure once attempts are exhausted.

retry_fixed(call_api, attempts=3, delay=1)

delay = min(base * (2 ** (attempt - 1)), cap)

Doubles the retry delay each attempt, capped at a maximum. Add random jitter (e.g. random.uniform(0, delay)) to avoid synchronized retries.

[compute_backoff(1.0, n, 30.0) for n in range(1, 8)] # [1.0, 2.0, 4.0, 8.0, 16.0, 30.0, 30.0]

logger.exception(message)

Called inside an except block; logs at ERROR level with the full traceback attached automatically. Equivalent to logger.error(message, exc_info=True).

except ValueError: logger.exception("risky_operation failed")

logger.error(detail); return {"error": generic_message}

Log full exception detail internally; return a short, generic message externally — internal detail exposed to a caller is an information leak.

except OrderError as e: logger.error("order failed: %s", e) return {"error": "We could not process your order."}

except Exception: pass # AVOID unless intentionally justified with a comment

Discarding an exception with no log, metric, or re-raise hides real failures. Log at minimum, and catch the narrowest type actually expected.

except ValueError as e: logger.error("skipping bad row: %s", e) continue

with expression as name: ...

Runs expression.__enter__() before the block and expression.__exit__() after — even if the block raises.

with open("data.txt") as f: contents = f.read()

from contextlib import contextmanager, suppress, closing, ExitStack

Standard library module for building and combining context managers without writing __enter__/__exit__ by hand.

with suppress(FileNotFoundError): open("missing.txt")

@contextmanager def name(arg): # setup try: yield value finally: # teardown

Turns a generator function into a context manager — code before yield is setup, code after yield (in finally) is teardown.

with my_manager("x") as value: ...

with a() as x, b() as y: ...

Enters a then b, left to right; exits b then a — last entered, first exited.

with open("in.txt") as fin, open("out.txt", "w") as fout: fout.write(fin.read())

async def __aenter__(self): ... async def __aexit__(self, exc_type, exc_value, traceback): ...

The async counterpart of __enter__/__exit__ — both coroutines, awaited by async with, for setup/teardown that itself awaits something.

async with AsyncConnection() as conn: await conn.query(...)

with acquire_resource() as thing: use(thing)

The reason to reach for a context manager: guaranteed cleanup for files, transactions, locks, connections, and temporary state.

with lock: process_shared_data() # released no matter what

x: int = 5 def f(a: str, b: int = 1) -> str: ...

Type annotation syntax for a variable, a function parameter, and a return value. Documentation only — not enforced at runtime.

def greet(name: str) -> str: return "Hi " + name

list[T] · dict[K, V] · tuple[T, ...] · set[T]

Built-in container generics (PEP 585). Work directly since Python 3.9 — no typing.List/Dict/Tuple/Set import needed.

scores: dict[str, int] = {"Ana": 90}

Optional[X] == X | None Union[X, Y] == X | Y

Optional[X] means X or None. Union[X, Y] means X or Y. The X | None / X | Y spelling (PEP 604, Python 3.10+) is the modern equivalent.

def get_user(user_id: int) -> User | None: ...

Any · Literal["a", "b"] · Final = value

Any disables checking for a value. Literal restricts to specific values. Final marks a name as not meant to be reassigned. None enforced at runtime.

MAX_RETRIES: Final = 3 Mode = Literal["r", "w"]

Callable[[ArgType, ...], ReturnType]

Annotates a value that is itself a function with a given signature. Use Callable[..., T] for "any arguments."

def apply(fn: Callable[[int, int], int]) -> int: return fn(1, 2)

T = TypeVar("T") class Box(Generic[T]): ...

TypeVar declares a placeholder type name; Generic[T] makes a class parametrizable over it.

b: Box[int] = Box(5) b.get() # type checker knows this is int

Name: TypeAlias = ExistingType Annotated[Type, metadata, ...]

TypeAlias names an existing type for reuse in signatures. Annotated attaches extra metadata to a type without changing the type itself.

UserId: TypeAlias = int PositiveInt = Annotated[int, "must be > 0"]

class Name(TypedDict): key: Type

Declares a dict with fixed, known keys and a type per key. Produces a plain dict at runtime — checked only by a static type checker.

class Movie(TypedDict): title: str year: int

if isinstance(x, T): ... # auto-narrows def guard(x) -> TypeGuard[T]: ... # custom

Type narrowing shrinks a Union to one branch after a check like isinstance(). A type guard is a custom function that teaches a type checker a narrowing rule.

if isinstance(value, str): value.upper() # narrowed to str

def f(x: list[T]) -> T: ... class C(Generic[T]): ...

A generic function reuses a TypeVar between parameter and return type. A generic class inherits Generic[T] and uses T across its methods.

def first(items: list[T]) -> T: return items[0]

TypeVar("T", covariant=True | contravariant=True)

Variance: whether a generic container built from a subtype may substitute for one built from its supertype. Covariant for read-only, contravariant for write-only, invariant (default) for both.

T_co = TypeVar("T_co", covariant=True) class ReadOnlyBox(Generic[T_co]): ...

mypy file.py npx pyright file.py

The two mainstream Python static type checkers. Both analyze annotations without running the code; mypy is the Python-native original, pyright (Pylance) is Microsoft's, generally faster.

pip install mypy mypy app.py

mypy --strict . # CI: pip install mypy && mypy .

IDE integration checks live as you type; --strict enables every optional check; running the checker in CI is what makes it mandatory for every change.

- run: pip install mypy && mypy .

@dataclass class Name: field1: type field2: type = default

Generates __init__, __repr__, and __eq__ from a class's annotated fields.

@dataclass class Order: order_id: str total_cents: int

@dataclass(frozen=True) class Name: ...

Blocks attribute assignment after construction — raises FrozenInstanceError on obj.field = value.

@dataclass(frozen=True) class Coordinates: lat: float lon: float

field(default_factory=list, init=True, repr=True, compare=True, metadata=None)

Configures one field beyond a plain default — required for any mutable default value.

items: list = field(default_factory=list)

def __post_init__(self): ...

Runs automatically as the last step of the generated __init__, after every field is assigned.

def __post_init__(self): if self.value < 0: raise ValueError("must be non-negative")

@dataclass(order=True) class Name: ...

Generates __lt__/__le__/__gt__/__ge__, comparing fields as a tuple in declaration order.

@dataclass(order=True) class Version: major: int minor: int

@dataclass class Sub(Base): ...

Inherits the parent dataclass's fields, combined into one __init__ — parent fields first, in order.

@dataclass class Dog(Animal): breed: str = "mixed"

@dataclass(slots=True) class Name: ...

Generates __slots__ from the field list (Python 3.10+) — no __dict__, lower memory, no undeclared attributes.

@dataclass(slots=True) class Point3D: x: float y: float z: float

@dataclass(frozen=True, slots=True) class Name: ... # value object

A regular class needs hand-written boilerplate; @dataclass generates it but never validates; Pydantic validates and coerces at construction. A value object is usually a frozen dataclass.

@dataclass(frozen=True) class Money: amount_cents: int currency: str

Counter(iterable) · defaultdict(factory) · namedtuple(name, fields) · deque(iterable, maxlen=None)

Four specialized containers for counting, auto-initializing, naming fields, and fast double-ended queues.

Counter(words).most_common(3)

chain(*its) · islice(it, stop) · groupby(it, key) · product/combinations(it, r)

Lazy building blocks for chaining, slicing, grouping, and generating combinations over iterators.

list(chain([1, 2], [3, 4]))

@lru_cache(maxsize=128) · reduce(fn, it, start) · partial(fn, *args) · @wraps(fn)

Caching, folding, argument pre-filling, and metadata preservation for functions.

@lru_cache(maxsize=None) def fib(n): ...

isinstance(x, collections.abc.Sequence)

Abstract base classes checking capability (Sequence, Mapping, Iterable) rather than one concrete type.

isinstance([1, 2, 3], collections.abc.Sequence)

Path(*parts) / "next" · .name/.stem/.suffix/.parent · .exists()/.read_text()/.glob(pattern)

Object-oriented filesystem paths — joining, reading pieces, and checking the filesystem.

Path("data") / "file.csv"

datetime(y, m, d, h, mi, s, tzinfo=ZoneInfo("Region/City")) · timedelta(days=, hours=)

Dates, times, durations, and real IANA time zones with correct DST handling.

datetime.now(ZoneInfo("Europe/London"))

os.environ.get(k, default) · sys.argv · subprocess.run([...], capture_output=True, text=True)

Environment variables and OS paths (os), interpreter state (sys), and running other programs (subprocess).

subprocess.run(["git", "status"], capture_output=True, text=True)

@contextmanager def name(): ... yield ...

Build a context manager from a generator function instead of a class — full depth in Context Managers.

@contextmanager def timer(): yield

async def f(): ... · await expr · asyncio.run(main()) · asyncio.gather(*coros)

Single-threaded concurrency for I/O-bound work — full depth in the Async Python section.

asyncio.run(asyncio.gather(fetch(a), fetch(b)))

with ThreadPoolExecutor(max_workers=N) as pool: pool.map(fn, items) / pool.submit(fn, *args)

A uniform pool API over threads (I/O-bound) or processes (CPU-bound).

list(pool.map(square, [1, 2, 3, 4]))

logging.getLogger(__name__).warning(msg, *args)

Leveled, filterable, routable output — DEBUG through CRITICAL, via loggers, handlers, and formatters.

logger.error("failed to save order %s: %s", order_id, exc)

re.search(pattern, s) · re.findall(pattern, s) · re.sub(pattern, repl, s)

Pattern matching, extraction, and replacement against strings.

re.sub(r"\s+", " ", text)

json.dumps(obj, indent=2, sort_keys=True) · json.loads(s)

Serialize Python objects to JSON text and parse JSON text back.

json.dumps({"active": True}, indent=2)

class X(Enum): NAME = value · X.NAME.value · list(X) · X(value)

A fixed, named, comparable set of values as a real type.

class Status(Enum): PENDING = "pending"

parser.add_argument("--flag", type=int, default=3, action="store_true")

Declarative command-line argument parsing with automatic --help.

parser.parse_args(["file.csv", "--verbose"])

@dataclass class Name: field: type

Auto-generates __init__/__repr__/__eq__ from annotated fields — full depth in Dataclasses and Data Modeling.

@dataclass class Point: x: int y: int

Decimal("19.99") · Fraction(1, 3) · statistics.mean(data)

Exact decimal arithmetic, exact rational arithmetic, and summary statistics — three numeric modules beyond plain float.

Decimal("0.1") + Decimal("0.2") == Decimal("0.3")

uuid.uuid4() · secrets.token_hex(n) · hashlib.sha256(data).hexdigest() · shutil.copy(src, dst)

Six single-purpose stdlib modules: embedded database, unique ids, secure randomness, scratch files, file operations, hashing.

hashlib.sha256(b"data").hexdigest()

concurrency: interleaved · parallelism: simultaneous

Concurrency structures work as multiple interleavable tasks; parallelism actually runs tasks at the same instant, which requires multiple cores.

threading.Thread(...) # concurrency multiprocessing.Process(...) # parallelism

CPU-bound -> multiprocessing · I/O-bound -> threading/asyncio

CPU-bound work is limited by computation and needs separate cores to speed up; I/O-bound work is limited by waiting and speeds up by overlapping those waits.

with ProcessPoolExecutor() as ex: ex.map(cpu_heavy, items) with ThreadPoolExecutor() as ex: ex.map(fetch_url, urls)

Thread/Process (one-off) · ThreadPoolExecutor/ProcessPoolExecutor (reused pool)

Four building blocks: a single thread or process for a one-off task, or a thread/process pool for many tasks of the same shape.

with ThreadPoolExecutor(max_workers=8) as ex: ex.map(fetch_url, urls)

async def f(): ... · await f() · asyncio.run(main())

A coroutine pauses only at an explicit await, letting the event loop run another coroutine on the same thread — a fit for many concurrent I/O waits.

results = await asyncio.gather(fetch("a"), fetch("b"))

threading.Thread(target=fn, args=(...)).start() / .join()

Builds and runs a thread with .start(), and blocks the caller until it finishes with .join().

t = threading.Thread(target=download, args=("file.csv",)) t.start() t.join()

with threading.Lock(): ... · with threading.RLock(): ...

Lock allows only one thread in at a time, including re-entry by its own holder (which deadlocks). RLock allows the owning thread to re-acquire.

lock = threading.Lock() with lock: counter += 1

Event().wait()/.set() · Condition().wait_for(pred) · Semaphore(n)

Event signals a one-off happening, Condition waits for shared state to become true, Semaphore(n) limits concurrent access to n holders.

sem = threading.Semaphore(2) with sem: do_limited_work()

queue.Queue() — thread-safe put()/get(), no external lock needed

A thread-safe structure handles its own internal locking; queue.Queue is the standard way to hand off work between threads safely.

q = queue.Queue() q.put(item) # from any thread item = q.get() # from any thread, blocks until available

race: lost update · deadlock: circular lock wait · starvation: perpetual denial

Three concurrency failure modes: unsynchronized access loses updates, circular lock waiting deadlocks, and unfair scheduling starves a thread.

first, second = sorted([lock_a, lock_b], key=id) with first: with second: ...

mp.Process(target=fn, args=(...)) · with mp.Pool(processes=n) as pool: pool.map(fn, items)

Runs work in real, separate OS processes — genuine CPU parallelism, but every argument and result crosses the process boundary by pickling.

with mp.Pool(processes=2) as pool: results = pool.map(cpu_heavy, [n1, n2])

mp.Queue() / mp.Pipe() · shared_memory.SharedMemory(create=True, size=n)

Queue/Pipe move data between processes by pickling it; shared_memory gives multiple processes a raw buffer with no pickling, at the cost of manual close()/unlink() bookkeeping.

shm = shared_memory.SharedMemory(create=True, size=10) shm.buf[0] = 42

with ThreadPoolExecutor(max_workers=n) as ex: ex.submit(fn, *args)

A unified pool interface — ThreadPoolExecutor for I/O-bound work, ProcessPoolExecutor for CPU-bound work — where submit()/map() return Futures instead of blocking immediately.

with ThreadPoolExecutor(max_workers=2) as ex: f = ex.submit(fetch_url, url) result = f.result()

future.result(timeout=None) · as_completed(futures) · future.cancel()

.result() blocks and re-raises the task's exception; as_completed() yields futures as they finish; cancel() only succeeds before the task starts running.

for future in as_completed(futures): print(future.result())

sys._is_gil_enabled()

The GIL is CPython's single mutex around Python bytecode execution, existing to keep reference counting safe without per-object locks.

import sys print(sys._is_gil_enabled()) # True on a standard build

CPU-bound → ProcessPoolExecutor · I/O-bound → ThreadPoolExecutor / asyncio

The GIL makes CPU-bound threading ineffective (sometimes slower than serial) but leaves I/O-bound threading genuinely effective, since blocking I/O releases the GIL.

Measured: CPU-bound 2 threads 0.757s vs. 2 processes 0.470s vs. serial 0.641s. I/O-bound 2 threads 0.302s vs. serial 0.601s.

async def f(): ... · await f() · asyncio.run(main())

async def builds a coroutine function; calling it returns a paused coroutine object; await is what actually runs it.

result = await greet("Ada")

asyncio.create_task(coro())

Schedules a coroutine to start running immediately, in the background, and returns an awaitable Task.

t = asyncio.create_task(worker(4)) result = await t

await asyncio.gather(*coros, return_exceptions=False)

Runs coroutines concurrently, returns results in call order once all finish.

results = await asyncio.gather(fetch("a"), fetch("b"))

task.cancel() · async with asyncio.timeout(seconds): ...

Both interrupt a coroutine at its next await point — cancel() on demand, timeout() automatically after a deadline.

async with asyncio.timeout(2): await slow_operation()

async def __aenter__(self) · async def __aexit__(self, exc_type, exc, tb)

The async equivalents of __enter__/__exit__, used with async with for resources whose open/close step is itself I/O.

async with AsyncDBConnection() as conn: await conn.query("SELECT 1")

async def gen(): ... yield x ... · async for x in gen(): ...

An async def function with yield is an async generator — Python builds __aiter__/__anext__ automatically.

async for v in countdown_gen(3): print(v)

sem = asyncio.Semaphore(n) · async with sem: ...

Caps how many coroutines can be inside the block at once; Lock() is the same idea with a cap of exactly 1.

async with asyncio.Semaphore(2): await fetch_page(n)

await asyncio.to_thread(blocking_fn, *args)

Runs a blocking function in a worker thread and awaits its result, without freezing the event loop.

result = await asyncio.to_thread(blocking_io)

async with httpx.AsyncClient() as client: await client.get(url)

An async HTTP client awaits network I/O instead of blocking the event loop; reuse one instance as a connection pool.

async with httpx.AsyncClient() as client: r = await client.get(url)

asyncio.Queue(maxsize=n)

A bounded queue whose put() blocks once full — forces a fast producer to wait for a slower consumer, applying backpressure automatically.

queue = asyncio.Queue(maxsize=100) await queue.put(item) # blocks if full

try: await work() finally: await cleanup()

try/finally guarantees cleanup runs on cancellation exactly like any other exception — the standard shape for graceful cancellation.

try: await do_work() finally: await resource.release()

Threading → I/O-bound sync · Multiprocessing → CPU-bound · AsyncIO → high-concurrency I/O-bound

The three concurrency tools, matched to the three workload shapes the roadmap calls out.

CPU-bound -> ProcessPoolExecutor I/O-bound (few) -> ThreadPoolExecutor I/O-bound (many) -> asyncio.gather

Unit → Integration → API → End-to-end

The test pyramid: fast and narrow at the base, slow and broad at the top — most coverage should be in the base.

def test_unit(): ... def test_integration(db): ... def test_api(client): ...

test_regression_<bug-id>_<what-it-guards>()

A regression test names the bug it guards against; a contract test checks response shape; a performance test asserts a time/resource budget.

def test_regression_1234_negative_quantity_returns_zero(): assert calculate_total(-1, 10) == 0

test_*.py · def test_*(): assert ...

pytest discovers tests by naming convention and rewrites plain assert statements to show exactly what failed.

def test_add_returns_sum(): assert add(2, 3) == 5

@pytest.fixture(scope="function") def name(): yield value

Marks reusable setup/teardown, injected into any test naming it as a parameter — scope controls how often it rebuilds.

@pytest.fixture def sample_order(): return {"id": 1} def test_x(sample_order): ...

@pytest.mark.parametrize("a,b", [(1,2), (3,4)])

Runs the decorated test once per tuple of values, each reported as its own separate test ID.

@pytest.mark.parametrize("a,b,expected", [(2,3,5)]) def test_add(a, b, expected): assert a + b == expected

@pytest.mark.skip/skipif/xfail · pytest -m "expr"

Marks tag tests for selection or special handling; -m filters the run by mark expression.

@pytest.mark.skipif(sys.platform == "win32", reason="posix-only") def test_x(): ...

conftest.py: @pytest.fixture def name(): ...

Fixtures in conftest.py are auto-discovered by every test file in the same directory tree, without an import.

# conftest.py @pytest.fixture def api_client(): return build_client()

Mock(return_value=X) · mock.assert_called_once_with(...)

A stand-in object that records calls and returns a configured value; MagicMock adds dunder method support.

mock_send = Mock(return_value=True) mock_send("a@b.com") mock_send.assert_called_once_with("a@b.com")

with patch("module.name") as mocked: ... · AsyncMock(return_value=X)

patch swaps a name for a Mock for a scoped duration; AsyncMock is the awaitable Mock variant for coroutines.

fetcher.fetch = AsyncMock(return_value={"ok": True}) await fetcher.fetch(url)

monkeypatch.setenv/setattr/setitem(...)

A built-in fixture for temporary changes that automatically reverse after the test, pass or fail.

def test_x(monkeypatch): monkeypatch.setenv("API_KEY", "test-key")

Stub (canned answer) · Spy (records + real work) · Fake (simplified real impl)

The vocabulary for test doubles by what they actually do, not just "mock" for everything.

class FakeRepository: def save(self, x): self._data[x["id"]] = x

Mock the boundary, not the thing under test

External dependencies (network, DB, payment, clock) are mock candidates; the function/class the test exists to check is not.

def test_places_order(mock_payment_gateway): place_order(cart, mock_payment_gateway)

pytest --cov=module --cov-report=term-missing

Reports the percentage of lines executed by the test suite, and exactly which lines were missed.

pytest --cov=calc --cov-report=term-missing

No shared mutable state · no dependence on real time/randomness/order

Isolated tests do not depend on other tests' side effects; deterministic tests give the same result every run.

def test_is_expired(): order = Order(created_at=datetime(2026, 1, 1)) assert order.is_expired(now=datetime(2026, 2, 1))

@pytest.fixture: begin transaction -> yield -> rollback

Wraps a test in a transaction that never commits, so its database writes are invisible to every other test.

@pytest.fixture def db_session(): tx = connection.begin() yield session tx.rollback()

with pytest.raises(ExceptionType, match="text"): ... · mock.side_effect = Error(...)

Asserts an error actually happens, and forces a mocked dependency to fail on demand to test error-handling paths.

with pytest.raises(ValueError, match="positive"): validate_amount(-10)

python -m venv .venv · pip install <package>

Creates an isolated environment, then installs a package into whichever environment is active.

python -m venv .venv .venv\Scripts\activate pip install requests

[build-system] · [project] · src/mypackage/

One TOML file for build config and metadata; src/ layout is the modern recommended directory shape.

[project] name = "mypackage" version = "0.1.0"

python -m build → dist/*.whl + dist/*.tar.gz

Runs the build backend to produce a wheel and a source distribution from pyproject.toml.

pip install build python -m build

pip install -e .

Links a project's source directly into the current environment — edits take effect immediately, no reinstall.

pip install -e . python -c "import mypackage"

pkg==1.2.3 (pin) · pkg>=1.2,<2.0 (range) · pkg~=1.2 (compatible release)

A pin locks to one exact version; a range accepts any version within a window.

dependencies = ["requests>=2.31,<3.0"]

MAJOR.MINOR.PATCH

PATCH = bug fix, MINOR = compatible new feature, MAJOR = breaking change.

2.31.0 -> 2.31.1 (patch) -> 2.32.0 (minor) -> 3.0.0 (major)

pip check

Verifies every installed package's declared requirements are actually satisfied by what is present.

pip install package-a package-b pip check

pip install --index-url <private-url> <package>

Installs from a private package registry instead of public PyPI.

pip install --index-url https://internal.example.com/simple/ mypackage

python -m build → twine check dist/* → twine upload dist/*

Build, validate, then upload artifacts to PyPI or a private index.

python -m build twine upload dist/*

setup.py (executed) vs. pyproject.toml (parsed, not executed)

pyproject.toml is static data read without running any code; setup.py is Python code pip must execute.

[project] name = "mypackage" version = "0.1.0"

python -m venv .venv (packages) · uv python install 3.12 (interpreter)

A venv isolates packages; a version manager isolates which Python interpreter is even available.

uv python install 3.12 uv venv --python 3.12

pip install . (prod) · pip install ".[dev]" (prod + dev)

[project.optional-dependencies] dev = [...] adds dev-only packages on top of the base dependencies.

[project.optional-dependencies] dev = ["pytest>=8.0", "ruff>=0.5"]

from dotenv import load_dotenv; load_dotenv()

Loads a .env file's NAME=value lines into os.environ — must run before reading any of those variables.

load_dotenv() db_url = os.environ.get("DATABASE_URL")

os.environ.get("API_KEY") → raise if missing

Read a secret from an environment variable and fail loudly and immediately if it is not set.

api_key = os.environ.get("API_KEY") if not api_key: raise RuntimeError("API_KEY must be set")

uv lock → uv.lock → uv sync

Resolves every dependency to an exact version and hash, then installs precisely what was locked.

uv lock uv sync

uv venv · uv pip install <pkg> · uv lock · uv sync

A faster, all-in-one alternative to pip + venv + a separate lock-file tool.

uv venv uv pip install requests

poetry add <pkg> · pip-compile requirements.in

Poetry is an all-in-one manager; pip-tools adds just a lock step to plain pip.

poetry add requests # or: pip-compile requirements.in

conda create -n myenv python=3.12 · pyenv install 3.12

Conda handles Python plus non-Python dependencies; pyenv manages multiple Python interpreter versions.

conda create -n myenv python=3.12 conda activate myenv

black <file> · isort <file> · black --check <file>

Automatically reformat code (black) and sort imports (isort) to a consistent style.

black myfile.py isort myfile.py

ruff check <file> · flake8 <file>

Both lint for bugs and style issues — Ruff is faster and all-in-one; Flake8 relies on separate plugins.

ruff check --fix myfile.py

pre-commit install → git commit (hooks run automatically)

Registers a real git hook that runs configured checks before every commit is allowed to complete.

pip install pre-commit pre-commit install

mypy <file> · pyright <file>

Both statically check type hints for errors, without running any code — same syntax, different output format.

mypy myfile.py pyright myfile.py

bandit myfile.py · bandit -r .

Scans source code for known vulnerability patterns — shell injection, hardcoded secrets — with severity ratings.

bandit -r . --severity-level high

def clear_name(typed: Arg) -> ReturnType: """Why, not what."""

Clear naming, type annotations, and a docstring explaining non-obvious behavior make code fast to review and maintain.

def calculate_shipping_cost(order: Order) -> Decimal: """Free above the $50 threshold."""

One function, one job, one reason to change

A cohesive function does one thing and is trivially testable in isolation; abstractions should follow at least two real concrete cases.

def validate_order(order): ... def charge_order(order, gateway): ...

Refactor: structure changes, behavior does not — tests prove it

Keep tests green before and after a refactor; never mix structural and behavioral changes in one commit.

assert calculate_total_v1(items) == calculate_total_v2(items)

O(1) < O(log n) < O(n) < O(n log n) < O(n²)

The shape of how an algorithm's cost grows as input size grows — smaller is better at scale.

x in a_set # O(1) x in a_list # O(n)

x in a_list # O(n) · x in a_set # O(1) average

List membership scans every item; set/dict membership hashes directly to the answer.

blocked_ids = set(blocked_id_list) # convert once if user_id in blocked_ids: ...

CPU · Memory · I/O · Network · Database · Serialization

Six distinct root causes of slowness, each needing a different fix — profile to find which one applies.

python -m cProfile -s cumulative myapp.py

timeit.timeit(stmt, setup="", number=1000000)

Runs a small snippet repeatedly and returns total time, averaging out single-run noise.

timeit.timeit("x in s", setup="s = {1,2,3}", number=10000)

cProfile.run("main()", "out.prof") → pstats.Stats("out.prof").sort_stats("cumulative")

Profiles a whole program's function calls, then analyzes and prints the results sorted by cost.

python -m cProfile -s cumulative myapp.py

tracemalloc.start() → get_traced_memory() → tracemalloc.stop()

Tracks real memory allocations and reports current/peak usage in bytes.

tracemalloc.start() current, peak = tracemalloc.get_traced_memory()

@functools.lru_cache(maxsize=None)

Caches a pure function's results by argument — repeated calls with the same arguments return instantly.

@lru_cache(maxsize=None) def fib(n): return n if n < 2 else fib(n-1) + fib(n-2)

WHERE id IN (...) · select_related()/prefetch_related() · connection pool

Fewer, larger round trips beat many small ones — watch for N+1 queries hidden inside a loop.

users = db.query("SELECT * FROM users WHERE id IN (?)", user_ids)

ProcessPoolExecutor (CPU-bound) · asyncio.gather (I/O-bound)

Parallelize genuinely CPU-bound work across cores; overlap I/O-bound waits on one thread with async.

with ProcessPoolExecutor() as ex: results = list(ex.map(fn, items))

Profile → find the real bottleneck → optimize → re-profile

Never optimize based on intuition alone — a profiler measures what actually dominates runtime.

python -m cProfile -s cumulative myapp.py

sys.getsizeof(obj) · gc.get_referrers(obj)

Measures real object size and finds what still references an object — the starting tools for a memory investigation.

gc.get_referrers(suspected_leaked_object)

sys.getsizeof(container) — the container only, not what it references

Lists (and dicts/sets) over-allocate to amortize growth — real size grows in jumps, not linearly.

lst = [] for i in range(1000): lst.append(i) # occasional resizes, not every append

sum(x for x in data) — no brackets, no intermediate list

Dropping list-comprehension brackets to a generator expression avoids building an intermediate list for any one-pass reduction.

total = sum(x * x for x in huge_dataset)

class Row: __slots__ = ("id", "value")

Worth adding to classes instantiated in bulk — the per-instance saving multiplies into a real total.

__slots__ = ("id", "value") # add for a class with 100,000+ instances

copy.copy(x) (shallow, cheap) vs. copy.deepcopy(x) (deep, costs real memory)

A deep copy duplicates every nested object — genuinely more memory than a shallow copy, which shares them.

shallow = copy.copy(x) # nested objects shared deep = copy.deepcopy(x) # nested objects duplicated

snapshot2.compare_to(snapshot1, "lineno")

Diffs two tracemalloc snapshots, ranking which lines allocated the most new memory between them.

snapshot1 = tracemalloc.take_snapshot() # ... suspected leaking code ... snapshot2.compare_to(snapshot1, "lineno")

gc.collect() · gc.garbage

Forces cycle collection now and reports how many objects were freed — useful as a memory-investigation signal.

freed = gc.collect() print(f"freed {freed} objects")

weakref.WeakValueDictionary()

A cache whose entries vanish automatically once nothing else references the value — the fix for a cache that should not keep everything alive.

cache = weakref.WeakValueDictionary() cache["key"] = some_object

@lru_cache(maxsize=1000) — bounded, not None

An unbounded cache (maxsize=None) grows forever with distinct arguments — bound it, or key on something that repeats.

@lru_cache(maxsize=1000) def compute(x): ...

[obj for obj in gc.get_objects() if isinstance(obj, SuspectedClass)]

Counts live instances of a suspected class right now — the concrete confirmation step in a memory investigation.

leaked = [o for o in gc.get_objects() if isinstance(o, Connection)] print(len(leaked))

logging.getLogger(__name__).setLevel(logging.INFO)

Gets a module-scoped logger and sets the minimum severity it will accept — a handler can filter further.

logger = logging.getLogger(__name__) logger.warning('disk usage at %d%%', 91)

logger.info(msg, extra={"field": value})

Attaches a custom named field to one log record, readable back via record.<field> in a formatter.

logger.info('order created', extra={'order_id': 4821})

logging.LoggerAdapter(logger, {"request_id": rid})

Wraps a logger so every call through it automatically carries the same extra context, like a request ID.

req_log = logging.LoggerAdapter(logger, {'request_id': rid}) req_log.info('processing')

tracer.start_as_current_span(name)

Starts a span and makes it the active context — any span started inside the with-block nests under it automatically.

with tracer.start_as_current_span('handle-request'): ...

GET /healthz (liveness) vs. GET /readyz (readiness)

Liveness answers "restart me or not"; readiness answers "route traffic to me or not" — never conflate the two.

def readiness(): return ({'status': 'ok'}, 200) if db_ok() else ({'status': 'unavailable'}, 503)

GET/HEAD/OPTIONS = safe · GET/PUT/DELETE/HEAD/OPTIONS = idempotent

The safety/idempotency matrix that decides whether a request is safe to auto-retry.

client.put(url, json=full_state) # idempotent, safe to retry as-is

from http import HTTPStatus

Stdlib enum giving every standard status code a name, int value, and phrase — use the name, not a bare literal.

return {'id': new_id}, HTTPStatus.CREATED

headers={'Authorization': f'Bearer {token}'}

Headers are case-insensitive metadata sent alongside a request/response — never put secrets in the URL instead.

client.get(url, headers={'Accept': 'application/json'})

cookie['name'] = value; cookie['name']['httponly'] = True

A cookie set with HttpOnly + Secure + SameSite is the baseline-safe way to hand a browser a session ID.

c['session_id']['samesite'] = 'Lax'

headers={'If-None-Match': etag}

A conditional request — the server returns 304 with no body if the resource has not changed since that ETag.

resp = client.get(url, headers={'If-None-Match': cached_etag}) resp.status_code # 304 if unchanged

Access-Control-Allow-Origin: <specific-origin>

The core CORS response header — a browser only lets cross-origin JS read the response if this matches its own origin.

return {'Access-Control-Allow-Origin': 'https://app.example.com'}

secrets.compare_digest(submitted_token, expected_token)

Constant-time CSRF token comparison — never use == for a security-sensitive token comparison.

if not secrets.compare_digest(form_token, session_token): abort(403)

html.escape(untrusted_value)

Converts HTML-significant characters to entities so untrusted input renders as inert text, not executable markup.

safe = html.escape(user_comment) return f'<p>{safe}</p>'

scheme + host + port = origin

The same-origin policy compares all three — differ in any one and the browser treats it as a different origin.

httpx.Client(verify=True) # default -- never disable cert verification in production

POST /orders/{id}/cancellation, not POST /cancelOrder

A non-CRUD action modeled as a sub-resource stays addressable (GET-able) later — a verb-shaped endpoint is a dead end.

routes[('GET', '/users/{user_id}/orders')] = list_orders_for_user

class Model(BaseModel): field: int = Field(gt=0)

Pydantic validates on construction and raises a field-level ValidationError — the standard shape for request/response contracts.

model.model_dump() # explicit, controlled serialization

{"error": {"code": "...", "message": "..."}}

One consistent error envelope across every endpoint — a client branches on code, reads message, never parses a stack trace.

return {'error': {'code': 'order_not_found', 'message': '...'}}, 404

?page=2&page_size=20&sort=-created_at&status=paid&q=dragon

One collection endpoint, four independent query-parameter mechanics — pagination, sorting, filtering, and search.

total_pages = (total_items + page_size - 1) // page_size

/v1/orders vs. /v2/orders

URL-path versioning is the most visible, cache-friendly strategy — only endpoints with an actual breaking change need a new version.

VERSIONED_HANDLERS = {'v1': get_order_v1, 'v2': get_order_v2}

headers={"Idempotency-Key": uuid4()}

A client-generated key that lets the server safely dedupe a retried non-idempotent (POST) request.

if key in seen: return seen[key] # else process and store

order.owner_id == user.id or user.is_admin

Object-level authorization — checked per action, against the SPECIFIC resource, not just "is this user logged in."

if not can_delete_order(user, order): return 403

model.model_json_schema()

Generates an OpenAPI-compatible JSON Schema directly from the same model used for request validation — one source of truth.

CreateOrderRequest.model_json_schema()['required']

Additive = safe · Removed/renamed/retyped/narrowed = breaking

The concrete test for whether an API change can ship on the existing endpoint or needs a new version.

estimated_delivery: str | None = None # safe: new, optional

r.set(key, value) / r.get(key) / r.incr(key) / r.incrby(key, n)

set/get store and retrieve a string value under a key; incr/incrby atomically update an integer-valued key.

r.set("page_views", 1) r.incrby("page_views", 5) r.get("page_views") # '6'

r.rpush(key, *values) / r.lpop(key) / r.lrange(key, start, stop)

rpush/lpush push onto the tail/head; lpop/rpop remove from the head/tail; lrange(key, 0, -1) reads the full list in order.

r.rpush("job_queue", "task:a") r.blpop("job_queue", timeout=1)

r.sadd(key, *m) / r.sismember(key, m) / r.zadd(key, {m: score}) / r.zrevrange(key, 0, n, withscores=True)

Sets track unique unordered members with O(1) checks; sorted sets add a per-member score and stay readable in score order.

r.zadd("leaderboard", {"bob": 250}) r.zrevrange("leaderboard", 0, 0, withscores=True)

r.hset(key, mapping={...}) / r.hget(key, field) / r.hgetall(key) / r.hincrby(key, field, n)

Hashes store field-value pairs under one key — the shape for "one object, several named fields" without JSON-encoding it into a string.

r.hset("user:1001", mapping={"logins": 4}) r.hincrby("user:1001", "logins", 1)

r.expire(key, seconds) / r.ttl(key) / r.persist(key) / r.set(key, value, ex=seconds)

expire/ttl set and read a countdown on a key; persist removes it; set(..., ex=) writes the value and TTL together.

r.set("otp:42", "914213", ex=300) r.ttl("otp:42") # 300

r.pipeline(transaction=True) → pipe.watch(key) / pipe.multi() / pipe.execute()

pipeline() batches commands in one round trip; transaction=True wraps them in MULTI/EXEC; watch(key) aborts with WatchError if the key changed first.

with r.pipeline() as pipe: pipe.watch("stock:sku9") ... pipe.multi(); pipe.set(...); pipe.execute()

r.publish(channel, msg) / pubsub.subscribe(channel) — r.xadd(stream, fields) / r.xreadgroup(group, consumer, {stream: ">"})

Pub/Sub broadcasts live with no history; Streams append a durable, ID-ordered log consumer groups can read and replay.

r.xadd("events:orders", {"order_id": "1001"}) r.xrange("events:orders", "-", "+")

r.set(key, token, nx=True, px=ttl_ms) / r.register_script(release_lua)(keys=[key], args=[token])

nx=True acquires only if absent; a Lua script checks the token and deletes atomically so you never release a lock you no longer hold.

token = str(uuid.uuid4()) r.set("lock:job:5", token, nx=True, px=5000)

r.get(key) → miss → compute value → r.set(key, value, ex=ttl_seconds)

Cache-aside: read from Redis first, fall back to the real lookup on a miss, and always write the cached copy with a TTL.

r.set("cache:user_profile:1001", value, ex=300)

@functools.lru_cache(maxsize=128)

Caches a function's return value per distinct argument, in the current process's memory, evicting least-recently-used entries once maxsize is reached.

@lru_cache(maxsize=128) def slow_square(x): return x * x

redis.Redis(host=..., port=6379).set(key, value, ex=seconds)

Connects to a Redis server and stores a key with an optional TTL, shared across every process that connects to the same server.

r = redis.Redis(host="localhost", port=6379) r.set("user:42", "Alice", ex=300)

cache-aside · write-through · write-back

The three patterns for when a cache is populated relative to a database write — lazily on read miss, together with the write, or write-now-flush-later.

if key in cache: return cache[key] value = db[key]; cache[key] = value # cache-aside

expires_at = time.monotonic() + ttl_seconds

A TTL cache stores an absolute expiry deadline at insert time and checks it on every read; invalidation instead removes the entry directly, tied to a write.

if time.monotonic() > expires_at: del store[key] # treat as a miss

with lock: if key in cache: return cache[key] ...rebuild...

Prevents a cache stampede by letting only the first thread for a key rebuild it, while others wait on the same lock and re-check before rebuilding again.

lock = locks.setdefault(key, threading.Lock()) with lock: if key in cache: return cache[key]

warm_cache_on_startup()

Pre-populates a cache with expected hot keys before serving traffic, avoiding a burst of simultaneous misses right after a deploy or restart.

def warm_cache_on_startup(): for key, value in load_top_products().items(): cache[key] = value

RedisCluster(host=..., port=...).set(key, value, ex=seconds)

A distributed cache client that routes each key to the correct node via consistent hashing, spreading data and load across multiple machines.

rc = RedisCluster(host="localhost", port=6379) rc.set("user:42", "Alice", ex=300)

strong consistency vs. eventual consistency

How closely a cached value tracks the source of truth — strong means always fresh (expensive), eventual means fresh within a bounded window (cheaper).

# write-through: near-strong db[key] = value; cache[key] = value

try: cache_get(key) except CacheUnavailable: return db[key]

Fail-open pattern: on a cache-specific error, fall back to the source of truth so the cache being down degrades latency, not correctness.

try: return cache_get(key) except CacheUnavailable: return db[key]

Celery(name, broker="redis://...") | Queue(connection=Redis()) | actor (dramatiq)

Celery and RQ both wrap a broker connection and a decorator that turns a function into a background task; Dramatiq uses an "actor" decorator with a similar shape.

app = Celery("tasks", broker="redis://localhost:6379/0") @app.task def my_task(x, y): return x + y

queue.put(job) → worker: job = queue.get() → process → queue.task_done()

A producer enqueues; a worker dequeues, runs, and acknowledges — the four stages every task queue library automates over a real broker.

work_queue.put({"id": 1}) job = work_queue.get() ... work_queue.task_done()

ThreadPoolExecutor(max_workers=N) | ProcessPoolExecutor(max_workers=N) | celery worker --concurrency=N

The number of jobs run at once — threads/async for I/O-bound jobs, processes for CPU-bound jobs.

with ThreadPoolExecutor(max_workers=4) as pool: results = list(pool.map(do_work, jobs))

delay = base_delay * (2 ** (attempt - 1))

Exponential backoff: each retry waits roughly twice as long as the last, up to a max_attempts cap that eventually gives up.

for attempt in range(1, max_attempts + 1): try: return fn() except RuntimeError: if attempt == max_attempts: raise time.sleep(base_delay * (2 ** (attempt - 1)))

if attempt == max_attempts: dead_letter_queue.append({"job": job, "error": str(exc)})

Route a job to a separate dead-letter queue once it exhausts its retries, instead of dropping it or retrying forever.

for attempt in range(1, max_attempts + 1): try: handler(job) break except ValueError as exc: if attempt == max_attempts: dlq.append({"job": job, "error": str(exc)})

task.apply_async(countdown=seconds) | task.apply_async(eta=datetime) | beat_schedule (cron-style)

countdown/eta delay a single call once; a beat/cron-style schedule repeats a task on an interval.

send_welcome_email.apply_async(args=[user_id], countdown=600)

if idempotency_key in processed: return processed[idempotency_key]

Check a stored idempotency key before doing the job's real work, so a redelivered (duplicate) job returns the original result instead of repeating a side effect.

def charge_card(idempotency_key, amount): if idempotency_key in processed_payments: return processed_payments[idempotency_key] result = {"status": "charged", "amount": amount} processed_payments[idempotency_key] = result return result

job_states[job_id] = {"status": ..., "error": ...}

Record a job's status and error on every transition, then filter for the failed state to find what needs recovery.

try: fn() job_states[job_id]["status"] = "succeeded" except Exception as exc: job_states[job_id]["status"] = "failed" job_states[job_id]["error"] = str(exc)

presentation -> service -> repository -> database

Layered stacks responsibility top-down; clean/hexagonal instead make every outer layer depend inward on the business rules through an interface.

class OrderService: def __init__(self, repository): self.repository = repository

value object (no identity) · entity (has identity) · aggregate (root + members)

DDD names the business's own concepts directly in code, and puts every rule about an aggregate behind its single root object.

order.add_line("sku-1", 2, unit_price) # raises inside the aggregate root if invalid

bus.publish(event_name, payload) · bus.subscribe(event_name, handler)

A publisher emits an event by name; any number of subscribed handlers run without the publisher referencing them.

bus.subscribe("order_placed", update_inventory) bus.publish("order_placed", {"order_id": 501})

def x_factory(kind): return {"a": A, "b": B}[kind]()

A Factory hides "which class?" behind one function; an Abstract Factory hides it for a whole matching family of classes.

notifier = notifier_factory("email") notifier.send("your order shipped")

Builder().field1(x).field2(y).build()

Each builder method sets one field and returns self; build() assembles and validates the finished object.

HttpRequestBuilder().method("POST").url("/orders").build()

class Adapter(Target): def __init__(self, adaptee): self.adaptee = adaptee

Adapter implements the interface callers expect and translates each call to the wrapped, incompatible object internally.

adapter = LegacyXmlAdapter(LegacyXmlParser()) print_items(adapter) # print_items only knows JsonDataSource

self.strategy = strategy; self.strategy.calculate(...)

The context class holds an interchangeable strategy object and delegates to it, instead of branching on a type flag itself.

cart = Cart(StandardShipping()) cart.shipping_strategy = ExpressShipping() # swap at runtime

subject.attach(observer) · subject.notify(event)

Observers register with a subject and are called back through a shared interface whenever the subject notifies — decoupled from each other and from the subject.

ticker.attach(PriceLogger()) ticker.set_price("ACME", 120)

class XCommand(Command): def execute(self): ... def undo(self): ...

A command bundles a request (and its reverse) as an object the invoker can store, run, and undo without knowing what it does.

remote.submit(TurnOnCommand(light)) remote.undo_last()

class AddOn(Base): def __init__(self, wrapped): self._wrapped = wrapped

A decorator wraps an object of the same interface and delegates to it, adding behavior before or after — stackable without one class per combination.

order = WithCaramel(WithMilk(Espresso())) order.cost() # 3.75

repo.get_by_id(id) · repo.add(entity)

A repository exposes get/add/list methods over an aggregate, hiding whether storage is SQL, an API, or memory behind one interface.

repo = InMemoryUserRepository() repo.add({"id": 1, "name": "Priya Shah"}) repo.get_by_id(1)

class XService: def __init__(self, repository): ... def do_the_use_case(self, ...): ...

A service class represents one use case and coordinates repositories/domain objects for it — the single place that logic lives.

service = OrderPlacementService(ProductRepository()) service.place_order("sku-1", 2)

def __init__(self, dependency): self.dependency = dependency

A class receives its dependencies as constructor arguments rather than constructing them itself — see the roadmap's own Dependency Injection section for the full treatment.

ReportGenerator(ConsoleLogger()).generate()

class Sub(ABC): @abstractmethod def method(self): ...

A high-level class should depend on an abstraction (an ABC or a duck-typed shape), never on one concrete implementation it builds itself.

class OrderService: def __init__(self, mailer: Mailer): self.mailer = mailer

def __init__(self, dependency): self.dependency = dependency

Accept a dependency as a constructor parameter and store it on self, rather than constructing it inside __init__.

service = UserService(Repository()) fake_service = UserService(FakeRepository())

def do_thing(data, dependency=default_dependency): return dependency(data)

Accept a dependency (often a function) as a parameter with a sensible default, so a caller can override it per call without editing the function.

total_price([10, 20, 30], flat_tax) send_email(to, subject, sender=fake_sender)

container.resolve("service_name")

Ask a container for a fully wired-up dependency by name; the container builds it and everything it depends on, caching the result.

container = Container() user_service = container.resolve("user_service")

def endpoint(dep=Depends(get_dependency)): ...

Mark a FastAPI endpoint parameter as a dependency to resolve before the endpoint runs; the endpoint receives the dependency's return value as a plain argument.

def read_items(session=Depends(get_db_session)): return query(session)

service = ClassUnderTest(FakeDependency())

Inject a fake implementation of a dependency in a test so the class under test runs without a real network call, database, or side effect.

fake_gateway = FakePaymentGateway() service = OrderService(fake_gateway) assert fake_gateway.charged == [49.99]

def fn(explicit_dependency, ...): ...

Pass every dependency a function actually uses as a parameter — a module-level global a function reads or mutates without declaring it cannot be swapped or tested in isolation.

def fetch_user(pool, user_id): return pool.fetch_user(user_id)

os.getpid() · os.getppid() · subprocess.run([...])

A process is an isolated, independently-numbered unit of execution; a thread runs inside one process and shares its memory.

import os print(os.getpid(), os.getppid())

signal.signal(signal.SIGTERM, handler)

Registers a Python function to run when the named signal arrives, replacing its default action. SIGKILL and SIGSTOP cannot be caught.

def handler(signum, frame): shutdown_requested = True signal.signal(signal.SIGTERM, handler)

os.chmod(path, 0o600) · Path(path).chmod(0o755) · stat.filemode(mode)

File permissions are three octal-encoded triads (owner/group/other) of read(4)/write(2)/execute(1); every file also has an owning user and group.

import stat stat.filemode(stat.S_IFREG | 0o600) # "-rw-------"

os.environ.get("KEY", default) · os.environ["KEY"] · PurePosixPath("/etc") / "app"

Environment variables carry per-process configuration; the filesystem follows a conventional layout (/etc config, /var/log logs, /tmp scratch, /home user files).

db_host = os.environ.get("DB_HOST", "localhost")

ps aux · top · htop · lsof -i :PORT · kill -9 PID

ps/top/htop inspect running processes and resource use; lsof lists a process's open files and network connections; kill sends it a signal.

subprocess.run(["lsof", "-i", ":8000"])

grep -c PATTERN file · awk -F',' '{print $2}' file · sed 's/OLD/NEW/g' file

grep filters lines by pattern, awk extracts/processes fields, sed rewrites text — chainable with a pipe for quick log analysis without leaving the shell.

grep ERROR app.log | awk '{print $1}'

curl -s URL · curl -sI URL · curl -o FILE URL · curl -w "%{http_code}" -o /dev/null URL

curl sends an HTTP request from the shell and is built for scripting around the response; wget defaults to resumable, retrying downloads to disk.

subprocess.run(["curl", "-s", "-w", "%{http_code}", "-o", "/dev/null", url])

ssh user@host · scp file user@host:/path/ · systemctl restart NAME.service · crontab -e

ssh/scp connect to and move files onto a remote Linux host; systemd keeps a service running and restarts it on crash; cron runs a command on a schedule.

subprocess.run(["ssh", "deploy@host", "systemctl", "is-active", "myapp.service"])

(ip_address, port)

A network endpoint is an (IP address, port) pair -- the IP finds the host, the port finds the process.

ipaddress.ip_address('192.168.1.10').is_private # True

socket.socket(family, type) -> bind/listen/accept (server) or connect (client) -> send/recv

The stdlib socket module's core object for one end of a TCP or UDP connection.

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect(('example.com', 80))

socket.SOCK_STREAM (TCP) vs socket.SOCK_DGRAM (UDP)

TCP guarantees ordered, complete delivery after a handshake; UDP sends immediately with no guarantees.

udp.sendto(b'ping', (host, port)) # no connect() needed

socket.gethostbyname(host) / socket.getaddrinfo(host, port)

Resolves a hostname to an IP address via the OS DNS resolver.

socket.gethostbyname('www.python.org') # '167.82.56.223'

TCP handshake -> TLS handshake (HTTPS only) -> HTTP request/response

HTTP is application-layer over TCP; HTTPS inserts a TLS handshake between the TCP connection and the first HTTP bytes.

ssl.create_default_context().wrap_socket(tcp_sock, server_hostname=host)

client -> reverse proxy / load balancer -> backend instance(s)

A reverse proxy forwards requests to backend server(s), hiding them from the client; a load balancer additionally distributes those requests across multiple instances.

nginx upstream block + proxy_pass, or HAProxy backend + balance roundrobin

forward proxy: client-configured intermediary · NAT: router-level address rewriting

A proxy is chosen by the client at the application level; NAT is transparent IP-address translation at the network boundary.

httpx.Client(proxy='http://proxy.internal:8080')

socket.settimeout(seconds)

Bounds how long a blocking socket call (connect, recv) waits before raising socket.timeout instead of hanging.

s.settimeout(5) s.connect((host, port)) # raises socket.timeout if too slow

WebSocket: Upgrade: websocket -> 101 · SSE: Content-Type: text/event-stream

WebSockets are full-duplex over an upgraded connection; SSE is a one-way, plain-HTTP event stream with built-in browser reconnect.

yield 'data: {"progress": 42}\n\n' # one SSE event

WSGI: app(environ, start_response) → [bytes] ASGI: async def app(scope, receive, send) → None

WSGI is one blocking function per request; ASGI is one async function per connection that awaits and sends messages, adding WebSocket and lifespan support.

async def app(scope, receive, send): await receive() await send({"type": "http.response.start", "status": 200, "headers": []}) await send({"type": "http.response.body", "body": b"ok"})

uvicorn <module>:<attribute> [--host H] [--port P] [--workers N] [--reload]

Runs an ASGI app. --reload for development; --workers for basic multi-process production use; combine with Gunicorn for full process management.

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

gunicorn -w N [-k WORKER_CLASS] [--bind ADDR] [--graceful-timeout S] module:app

Forks N worker processes and supervises them. Default worker class is sync WSGI; use uvicorn.workers.UvicornWorker to run an ASGI app.

gunicorn -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 main:app

-w N (processes) · --worker-class {sync|gthread|UvicornWorker} · --threads N (gthread only)

Processes give real parallelism across CPU cores; threads help concurrent I/O-bound work within a process; an async worker's event loop handles many I/O-bound connections without extra threads.

gunicorn -w 9 -k uvicorn.workers.UvicornWorker main:app # (2*4 cores)+1, async worker

uvicorn --proxy-headers --forwarded-allow-ips <proxy-ip>

Puts Nginx/a load balancer in front of the app server for TLS termination, static files, and client buffering; trust forwarded headers only from that proxy's address.

uvicorn main:app --proxy-headers --forwarded-allow-ips 10.0.0.5

lifespan.startup / lifespan.shutdown · SIGTERM → drain (≤ graceful-timeout) → SIGKILL

Open shared resources on lifespan.startup, close them on lifespan.shutdown. A stop signal drains in-flight requests up to a configured timeout before the process is force-killed.

@asynccontextmanager async def lifespan(app): app.state.pool = await create_pool() yield await app.state.pool.close()

metrics / logs / traces → alerts, dashboards, SLOs, incident response

The three raw observability signals this section's tools (alerts, dashboards, SLOs, incident response) all consume — produced in Python as covered by Logging and Observability.

GET /healthz, GET /readyz

The first check an on-call engineer or an uptime monitor makes — full liveness/readiness mechanics covered in Logging and Observability.

IF metric compares-to threshold FOR duration THEN notify(channel)

The rule shape shared by Prometheus Alertmanager, CloudWatch Alarms, and Grafana alerting — a dashboard charts the same metric without the threshold or the page.

all(sample > threshold for sample in recent_samples)

error_budget = 1 - slo_target

SLI is the measured number, SLO is the target, error budget is 100% minus the SLO — negative remaining budget means the SLO is breached.

budget_remaining = error_budget - (1 - sli)

detect -> declare -> mitigate -> resolve -> analyze

The incident lifecycle, in order — mitigation always comes before full root cause analysis, which is modeled as a chain of contributing factors, not one cause.

if not mitigated: return "mitigate now"

summary, impact, timeline, root cause, what went well, action items (owner + status)

The standard blameless-postmortem sections from Google's SRE book — action items need an owner or the postmortem produces no real change.

[a for a in actions if a["status"] != "done"]

try: return call_dependency() / except SpecificError: return fallback

Catch the specific dependency exception and return a fallback value — a circuit breaker adds "stop calling after N consecutive failures" on top of the same idea.

if self.failure_count >= self.failure_threshold: self.open = True

if error_rate_breached(errors, total): return previous_version

A rollback reverts to the last known-good version — the fastest incident mitigation, since it does not require knowing the root cause first.

choose_active_version("v2.4.0", "v2.3.1", errors=80, total=1000) # -> "v2.3.1"

httpx.Client(timeout=httpx.Timeout(connect=2.0, read=5.0))

Bounds how long a single call may run before raising instead of hanging — set connect and read separately.

try: r = client.get(url) except httpx.TimeoutException: ... # decide next step yourself

time.sleep(random.uniform(0, min(cap, base * 2 ** (attempt - 1))))

Exponential backoff with jitter: wait grows each attempt, capped, and randomized within that cap.

for attempt in range(1, max_attempts + 1): try: return call() except TRANSIENT: if attempt == max_attempts: raise time.sleep(random.uniform(0, min(cap, base * 2 ** (attempt - 1))))

cb.call(downstream_fn, *args)

Routes a call through closed/open/half-open state — rejects immediately while open instead of attempting a known-failing downstream.

try: result = cb.call(fetch_from_downstream) except CircuitOpenError: result = fallback_value()

ClientSideLimiter(capacity, refill_rate).allow()

A client-side token bucket that self-throttles outgoing calls to protect a downstream, rather than the server protecting itself.

if limiter.allow(): call_downstream() else: queue_or_skip()

if dedup_key in processed: return processed[dedup_key]

The check that makes a retried write safe: recognize an already-completed request by its key instead of repeating the side effect.

result = processed.get(key) or run_and_store(key)

pool.acquire() # raises when every connection is checked out

A finite, shared resource — what happens on exhaustion (fail fast vs. block) is a reliability decision, not an afterthought.

try: conn = pool.acquire() except TimeoutError: return degrade_gracefully()

if len(value) > MAX: raise ValueError(...)

A field-length or list-length ceiling enforced alongside type validation, so a well-typed request cannot still be unbounded.

if len(items) > MAX_ITEMS: raise ValueError(f"items exceeds limit of {MAX_ITEMS}")

Bulkhead(name, capacity).run(fn)

An isolated, per-dependency resource pool — exhausting one bulkhead does not affect any other dependency's separate pool.

reports_pool = Bulkhead("reports", capacity=5) checkout_pool = Bulkhead("checkout", capacity=20)

except (ConnectionError, TimeoutError): return fallback_value()

Catch only the specific downstream-failure types and substitute a worse-but-working result, so the request still succeeds.

try: return live_value() except (ConnectionError, TimeoutError): logger.warning("using fallback") return fallback_value()

json.dumps(obj, indent=2) / json.loads(text)

Converts between Python objects and JSON text; raises TypeError on dumps for unsupported types, JSONDecodeError on loads for malformed text.

json.dumps({"user_id": 4821}, indent=2)

pickle.dumps(obj) / pickle.loads(data)

Serializes almost any Python object to bytes; loads can execute arbitrary code — never call on untrusted data.

blob = pickle.dumps({"user_id": 4821, "roles": {"admin"}})

csv.DictWriter(file, fieldnames=[...]) / csv.DictReader(file)

Reads and writes CSV rows as dicts keyed by column name, correctly quoting/unquoting fields that contain the delimiter.

list(csv.DictReader(open("users.csv")))

yaml.safe_dump(obj) / yaml.safe_load(text)

Third-party (PyYAML) round trip between Python objects and YAML text — always use the safe_ variants, never plain load/dump on untrusted input.

yaml.safe_load(open("config.yaml"))

MessagePack (no schema) · Protocol Buffers (.proto, numbered fields) · Avro (JSON schema, name-resolved)

Three binary serialization formats outside the stdlib — pick by whether you need a schema at all, and if so, whether cross-language RPC or pipeline-style schema evolution matters more.

len(json.dumps(obj).encode()) vs. len(pickle.dumps(obj))

Serialization cost is CPU (encode/decode time) plus size (bytes produced) — both are measurable on your actual payload, never assumed.

len(json.dumps(payload).encode("utf-8"))

record.get("new_field", default) — the read-side pattern for a safely evolved schema

Backward compatible = a new reader can still read old data. Add optional fields with defaults; avoid removing or renaming fields a live reader depends on.

sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("name").set_defaults(func=handler)

Split one CLI into named subcommands, each with its own scoped arguments and handler function.

args = parser.parse_args(["migrate", "--target", "0007"]) args.func(args)

os.environ.get("KEY", default) · sys.exit(0 | nonzero)

Read configuration from the environment safely; report success or failure through the process exit code.

if api_key is None: sys.exit(1)

subprocess.run(cmd_list, check=True, capture_output=True, text=True)

Run an external command, raise on failure, and capture its output as text — the automation-safe defaults.

try: subprocess.run(["deploy.sh"], check=True) except subprocess.CalledProcessError as e: sys.exit(e.returncode)

sys.exit(0 if success else 1) # the only thing cron reads

A cron-safe script logs instead of printing, is idempotent, and signals outcome only through its exit code.

logging.exception("job failed") sys.exit(1)

processed, errors = 0, 0 # track a summary; exit_code = 0 if errors == 0 else 1

A maintenance script logs each action, supports --dry-run, and is idempotent — safe to run again after a partial failure.

logger.info("removed %s", item) # not print() -- nobody watches an unattended run

for batch in chunked(fetch_records(), size=500): write_batch([migrate(r) for r in batch])

Transform existing data in batches, with a dry-run pass and a post-migration verification step.

for batch in chunked(items, size=500): process_batch(batch)

Process a large collection in fixed-size, independently retriable chunks instead of all at once.

batches = list(chunked(range(1, 11), 3)) # [[1,2,3], [4,5,6], [7,8,9], [10]]

subprocess.run(step, check=True) # gate on exit code, not printed text

Chain deployment or dev-tooling steps, stopping at the first non-zero exit code rather than parsing output.

for step in [build_cmd, push_cmd, deploy_cmd]: subprocess.run(step, check=True)

open(path, "r", encoding="utf-8") · open(path, "rb") · Path(path).read_text(encoding=...) · Path(path).read_bytes()

"r"/"w" (text) decode/encode using an explicit encoding and return/accept str; "rb"/"wb" (binary) return/accept raw bytes with no encoding involved.

Path("notes.txt").read_text(encoding="utf-8")

for line in open(path): ... · while chunk := f.read(size): ...

Iterate a file line by line, or read it in fixed-size chunks, to process it without loading the whole file into memory at once.

with open(path, "rb") as f: while chunk := f.read(8192): process(chunk)

tempfile.TemporaryDirectory() · tempfile.NamedTemporaryFile(delete=False)

TemporaryDirectory() auto-deletes a scratch directory on with-block exit. NamedTemporaryFile(delete=False) keeps the file on disk after closing, for handing its path to another process or handle.

with tempfile.TemporaryDirectory() as td: scratch = os.path.join(td, "work.txt")

Path(p).chmod(0o644) · stat.S_IMODE(Path(p).stat().st_mode)

Set a file's permission bits with chmod(); read them back with stat.S_IMODE() to mask off the file-type bits st_mode also encodes.

stat.S_IMODE(Path("file.txt").stat().st_mode)

gzip.open(path, "wb"/"rt") · zipfile.ZipFile(path, "w") · tarfile.open(path, "w:gz")

gzip compresses one stream; zipfile/tarfile bundle multiple files into one archive, with tarfile's "w:gz" mode combining bundling and compression.

with tarfile.open("bundle.tar.gz", "w:gz") as tf: tf.add("notes.txt")

with open(path) as f: ...

with guarantees a resource's cleanup (__exit__, which closes a file) runs on every exit path from the block — success, early return, or exception.

with open(path) as f: data = f.read()

date(y, m, d) · time(h, mi, s) · datetime(y, m, d, h, mi, s) · timedelta(days=, hours=)

The four core datetime module types — a day, a clock reading, a combined instant, and a duration.

datetime.combine(date(2026, 8, 21), time(9, 30)) + timedelta(hours=1)

dt.astimezone(ZoneInfo("Region/City"))

Converts an aware datetime to a different named time zone without changing the instant it represents.

datetime.now(timezone.utc).astimezone(ZoneInfo("Europe/London"))

dt.tzinfo is None → naive · dt.tzinfo is not None → aware

Naive datetimes carry no time zone and cannot be safely compared to aware ones — always construct aware datetimes with an explicit zone.

datetime.now(timezone.utc) # aware, not datetime.now()

datetime(..., fold=0|1, tzinfo=ZoneInfo(...))

fold disambiguates the repeated wall-clock hour created by a fall-back DST transition — fold=0 is earlier, fold=1 is later.

datetime(2026, 11, 1, 1, 30, fold=1, tzinfo=ZoneInfo("America/New_York"))

dt.isoformat() / datetime.fromisoformat(s) · dt.timestamp() / datetime.fromtimestamp(ts, tz=timezone.utc)

The two standard round-trip formats for a datetime — ISO 8601 text and Unix timestamp numbers.

datetime.fromtimestamp(dt.timestamp(), tz=timezone.utc) == dt

requests.get(url) | httpx.get(url) | await session.get(url)

requests: sync only. httpx: sync (Client) and async (AsyncClient) in one library, plus HTTP/2. aiohttp: async only.

async with httpx.AsyncClient() as client: r = await client.get(url)

HTTPAdapter(max_retries=Retry(total=3, status_forcelist=[...])), timeout=(connect, read)

Pool by reusing one Session/Client; bound wait time with an explicit timeout; retry via a mounted adapter (requests) or custom logic for status codes (httpx/aiohttp).

session.mount('https://', HTTPAdapter(max_retries=Retry(total=3))) session.get(url, timeout=(3.05, 10))

headers={'Authorization': f'Bearer {token}'} | auth=(user, pass) | auth=httpx.BasicAuth(user, pass)

Bearer auth is a plain header on any client; Basic auth has a dedicated auth= parameter per library.

session.headers.update({'Authorization': f'Bearer {token}'}) # applies to every request

requests: iter_content(N) | httpx: iter_bytes(N) | aiohttp: content.iter_chunked(N)

Read a response body in fixed-size chunks instead of loading it whole — needed for large downloads.

with requests.get(url, stream=True) as r: for chunk in r.iter_content(8192): f.write(chunk)

async with httpx.AsyncClient() as client: await asyncio.gather(*(client.get(u) for u in urls))

Reuse one AsyncClient/ClientSession and gather() several requests to run them concurrently — total time approaches the slowest single response, not the sum.

await asyncio.gather(*(client.get(u) for u in urls), return_exceptions=True)

response.raise_for_status()

No-op on 2xx; raises HTTPError (requests) or HTTPStatusError (httpx) on 4xx/5xx, with .response attached.

try: response.raise_for_status() except requests.exceptions.HTTPError as e: log(e.response.status_code)

response.json() # raises JSONDecodeError on malformed body

A 2xx status does not guarantee a valid or expected-shape body — parse defensively and check required fields.

try: data = response.json() except requests.exceptions.JSONDecodeError: handle_bad_response()

httpx.Client(proxy="http://p:8080") · httpx.Client(verify="/path/ca.pem")

Route a client through a proxy with proxy=/proxies=; trust a specific CA bundle with verify= rather than disabling certificate verification with verify=False.

client = httpx.Client(verify="/etc/ssl/certs/internal-ca.pem")

ADR: Context -> Decision -> Consequences

An architecture review evaluates a written ADR before implementation, and a "blocker:" concern from any reviewer gates approval.

review.raise_concern("priya", "blocker: no plan for message ordering guarantees")

def f(...) -> T: """What it does, what it raises, why (not obvious from code)."""

A docstring earns its keep by stating a contract the signature does not already make obvious; a README gets a new engineer running.

function_doc("charge_card", ["customer_id: str"], "a Charge record", ["CardDeclinedError"])

log.append({"correlation_id": correlation_id, "step": "...", ...})

Thread a correlation ID through every downstream call so a production request's full log timeline can be reconstructed after the fact.

isolated = [e for e in log if e["correlation_id"] == failing_cid]

(optimistic + 4 * most_likely + pessimistic) / 6

PERT three-point estimation weights the most-likely case heavily while still accounting for best and worst case.

three_point_estimate(3, 8, 21) # -> 9.3

task.get("behind_flag") or task.get("additive")

A subtask is safe to ship on its own if it is additive or hidden behind a feature flag — save the risky cutover for last.

shippable = [t for t in subtasks if is_independently_shippable(t)]

re.search(r"#\s*(TODO|FIXME|HACK|XXX)", line)

A debt-marker scan finds tracked, commented shortcuts — a starting point, not the whole picture of a codebase's technical debt.

find_debt_markers(source_lines) # -> [(2, "# HACK: ..."), (6, "# TODO: ...")]

observed = legacy_func(*args); assert observed == <actual current output>

A characterization test pins down what legacy code currently does (bugs included) so a later refactor has something real to check against.

assert legacy_discount(19.99, 10) == 179.9 # documents real behavior, not "correct" behavior

"new_service" if path in migrated_paths else "legacy_service"

A strangler-fig router sends already-migrated paths to the new system and leaves everything else on the legacy one.

strangler_route("/api/v2/invoices") # -> "new_service"

class MigrationNNNN: revision = "N" down_revision = "N-1" def upgrade(): ... def downgrade(): ...

A versioned schema migration chains to the prior revision and pairs every upgrade() with a genuine downgrade() inverse.

apply_migration(Migration0007AddLastLogin, "downgrade")

next_canary_step(percent, errors, total) -> new_percent

Widen canary traffic gradually while error rate stays under threshold; any breach reverts to 0% traffic immediately.

next_canary_step(50, errors=40, total=500) # -> 0 (breach)

review_comment(kind, text, teach=None)

A review comment that states the underlying "why" alongside a correction teaches a reusable lesson, not just a one-line fix.

review_comment("blocking", "mutable default", teach="evaluated once at def time")

compare_options([{name, gain, cost, recommended}, ...])

Present each option as gain versus cost in plain language, and mark an explicit recommendation rather than leaving the decision open.

compare_options(options) # "-> Add Redis cache: gain=...; cost=..."

def f(x): ... # x is bound to the argument object, like x = argument

Python has one binding rule, not two calling conventions: a parameter name becomes another name for the argument object. Reassigning the parameter never reaches the caller; mutating the object in place does.

def f(x): x.append(1) # caller sees this x = [] # caller does not see this

choose_model(cpu_bound, concurrent_operations) -> "multiprocessing" | "asyncio" | "threading"

CPU-bound work always needs multiprocessing regardless of concurrency scale; I/O-bound work needs asyncio only once concurrency scale is large, otherwise threading is simpler.

choose_model(cpu_bound=False, concurrent_operations=5000) # -> "asyncio"

reverse proxy -> WSGI/ASGI server -> worker -> framework routing -> view function

A request crosses four distinct hops before application code runs — naming all four, not just "nginx forwards to my app," is the complete answer.

HOPS = ["reverse proxy", "WSGI/ASGI server", "worker", "framework view"]

index = sorted(column) -> row pointer · log2(n) seek · one write per index per row change

An index trades write cost for read speed, and only pays off on a selective predicate. A condition matching 95% of the table is answered faster by scanning.

100,000 rows: full scan 99,999 comparisons, index lookup 17

atomic = all or nothing · isolation = what you see of concurrent transactions

Atomicity protects you from your own crash; isolation protects you from other transactions. A read-modify-write loses updates under Read Committed — use a relative or conditional update.

Both read 100, one writes 70, the other writes 50 -> the first withdrawal is gone

has_deadlock_cycle({tx_id: blocked_on_tx_id}) -> bool

A deadlock is a wait-for cycle between transactions, almost always from opposite-order row locking — fixed by always locking shared rows in a consistent order.

has_deadlock_cycle({"A": "B", "B": "A"}) # -> True

1. index? 2. N+1 (check query COUNT)? 3. SELECT */no LIMIT? 4. sargable WHERE? 5. EXPLAIN ANALYZE

A fixed diagnostic order for a slow query — index existence and hidden N+1 patterns explain most cases before a query plan is ever needed.

len(queries_executed) # 6 (N+1) vs 1 (JOIN) for the same 5 orders

document = assemble at write time · relational = assemble at read time

Pick by access patterns: one aggregate read whole favours documents, many cross-cutting questions favour relational. Embedding duplicates values, and every copy must be updated.

One order: 1 read vs 2. Revenue by product: 800 items touched vs 600. Rename: 200 copies vs 1 row.

should_use_redis(needs_durability, needs_relational_queries) -> bool

Redis fits caching, ephemeral data, and its native structures — not durable systems of record or relational queries.

should_use_redis(needs_durability=True, needs_relational_queries=False) # -> False

store.append(f"cart:{user_id}", item) # external store, not a module-level dict

A horizontally scalable Python service externalizes every piece of per-user state into a shared store every worker process can reach.

shared_store.append("cart:user-1", "widget") # any worker, any machine, sees this