← Back to the 16 Patterns

Advanced Python

The layer above the toolkit. Generators, decorators, context managers, the data model, and the concurrency story. What a senior Python interview probes, and what a code review will catch.

Guide 1 covers the containers and their costs, which is enough to solve the sixteen patterns. This page covers the language features that separate working Python from Python a reviewer would approve. Every example runs.

Iterators and generators

def countdown(n: int):
    """A generator function. `yield` makes it lazy.

    >>> list(countdown(3))
    [3, 2, 1]
    """
    while n > 0:
        yield n              # O(1) per item, and the function PAUSES here
        n -= 1


def flatten(groups: list[list[int]]):
    """`yield from` delegates to another iterable.

    >>> list(flatten([[1, 2], [3]]))
    [1, 2, 3]
    """
    for group in groups:
        yield from group     # equivalent to: for item in group: yield item


# A generator EXPRESSION: round brackets instead of square.
squares = (x * x for x in range(1_000_000))   # O(1) memory, nothing computed yet
first_square = next(squares)                  # O(1), computes exactly one value

Why it matters

from itertools import islice


def integers_from(start: int):
    """An INFINITE generator. Perfectly safe, as long as you stop taking.

    >>> list(islice(integers_from(10), 3))
    [10, 11, 12]
    """
    current = start
    while True:
        yield current
        current += 1


def running_total(values):
    """A pipeline stage: consumes an iterable, produces an iterable.

    >>> list(running_total([1, 2, 3]))
    [1, 3, 6]
    """
    total = 0
    for value in values:
        total += value
        yield total          # O(1) memory no matter how long the input is

Closures, and the late-binding trap

def make_adders_wrong() -> list:
    """Every lambda captures `i` itself, so all three see its FINAL value.

    >>> [f(10) for f in make_adders_wrong()]
    [12, 12, 12]
    """
    return [lambda x: x + i for i in range(3)]


def make_adders_right() -> list:
    """A default argument is evaluated at DEFINITION time, freezing the value.

    >>> [f(10) for f in make_adders_right()]
    [10, 11, 12]
    """
    return [lambda x, offset=i: x + offset for i in range(3)]

Scope: the LEGB rule

A bare name is looked up in four places, in order: Local, Enclosing function, Global (module), Built-in.

Decorators, written yourself

import functools
import time


def timed(func):
    """Wrap a function so its last runtime is recorded."""

    @functools.wraps(func)      # copies __name__, __doc__ and __wrapped__ over
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            wrapper.last_seconds = time.perf_counter() - start   # runs even on error
    return wrapper


@timed
def work(n: int) -> int:
    """Sum 0 to n-1.

    >>> work(5)
    10
    >>> work.__name__          # preserved by functools.wraps
    'work'
    """
    return sum(range(n))       # O(n)

A decorator that takes arguments

Then you need one more layer: a function that returns a decorator that returns a wrapper.

import functools

attempts = 0


def retry(times: int):
    """A decorator FACTORY. retry(3) evaluates to the actual decorator."""

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last: Exception | None = None
            for _ in range(times):
                try:
                    return func(*args, **kwargs)
                except ValueError as err:
                    last = err
            assert last is not None
            raise last
        return wrapper

    return decorator


@retry(3)
def flaky() -> str:
    """Fails twice, then succeeds.

    >>> flaky()
    'ok'
    """
    global attempts
    attempts += 1
    if attempts < 3:
        raise ValueError("not ready yet")
    return "ok"

Context managers

import time
from contextlib import contextmanager


class Timer:
    """A context manager as a class.

    >>> with Timer() as t:
    ...     total = sum(range(100))
    >>> t.seconds >= 0
    True
    """

    def __enter__(self) -> "Timer":
        self._start = time.perf_counter()
        return self                     # this is what `as t` receives

    def __exit__(self, exc_type, exc, traceback) -> bool:
        self.seconds = time.perf_counter() - self._start
        return False                    # False: do NOT swallow an exception


@contextmanager
def swallow_value_errors():
    """The same contract from a generator, which is usually shorter.

    Everything before the yield is __enter__. Everything after is __exit__.

    >>> with swallow_value_errors():
    ...     raise ValueError("handled")
    >>> "still running"
    'still running'
    """
    try:
        yield
    except ValueError:
        pass

The data model: dunder methods

class Deck:
    """Implement the protocols and the language does the rest.

    >>> deck = Deck([1, 2, 3])
    >>> len(deck)
    3
    >>> deck[0]
    1
    >>> 2 in deck
    True
    >>> [card for card in deck]
    [1, 2, 3]
    >>> deck
    Deck([1, 2, 3])
    """

    def __init__(self, cards: list[int]) -> None:
        self._cards = cards

    def __len__(self) -> int:
        return len(self._cards)         # O(1)

    def __getitem__(self, index: int) -> int:
        return self._cards[index]       # O(1); this ALONE gives iteration and `in`

    def __repr__(self) -> str:
        return f"Deck({self._cards!r})"  # !r calls repr on the inner value

dataclass options worth knowing

from dataclasses import dataclass


@dataclass(frozen=True, slots=True, order=True)
class Version:
    """frozen: immutable and hashable. slots: less memory. order: comparisons.

    >>> sorted([Version(1, 2), Version(1, 0)])
    [Version(major=1, minor=0), Version(major=1, minor=2)]
    >>> Version(1, 0) in {Version(1, 0)}
    True
    """

    major: int
    minor: int

Names, mutability and copying

import copy

original = [1, [2, 3]]

same = original                 # O(1)  a second NAME, not a copy
shallow = original.copy()       # O(n)  new outer list, SAME inner list
deep = copy.deepcopy(original)  # O(size) everything duplicated

original[1].append(4)
# same[1]    -> [2, 3, 4]   it is literally the same object
# shallow[1] -> [2, 3, 4]   the inner list was shared
# deep[1]    -> [2, 3]      fully independent

Passing arguments

def rebind(items: list[int]) -> None:
    items = [9, 9]         # rebinds the LOCAL name only. Caller sees nothing.


def mutate(items: list[int]) -> None:
    items.append(9)        # mutates the shared object. Caller DOES see this.

Python is neither “by value” nor “by reference”. The reference is passed by value, so rebinding is invisible to the caller and mutation is not. If a function mutates its argument, say so in the docstring, as every in-place solution on this site does.

Sorting, properly

records = [("bob", 3), ("ann", 3), ("cid", 1)]

by_count = sorted(records, key=lambda r: r[1])              # O(n log n)
by_count_then_name = sorted(records, key=lambda r: (r[1], r[0]))
count_desc_name_asc = sorted(records, key=lambda r: (-r[1], r[0]))

# When the descending key is NOT numeric, negating is impossible.
# Use two stable passes, least significant key FIRST.
step_one = sorted(records, key=lambda r: r[0])                    # name ascending
two_pass = sorted(step_one, key=lambda r: r[1], reverse=True)     # count descending

When a key function is not enough

from functools import cmp_to_key


def largest_number(numbers: list[int]) -> str:
    """Arrange the numbers to form the largest possible value.

    No key function can express this: the order depends on comparing PAIRS,
    because "3" before "30" is decided by "330" versus "303".

    >>> largest_number([3, 30, 34, 5, 9])
    '9534330'
    """

    def compare(a: str, b: str) -> int:
        if a + b > b + a:
            return -1            # a should come first
        if a + b < b + a:
            return 1
        return 0

    ordered = sorted((str(n) for n in numbers), key=cmp_to_key(compare))
    joined = "".join(ordered)
    return "0" if joined[0] == "0" else joined      # all zeros collapse to "0"

Exceptions and EAFP

counts = {"a": 1}
if "a" in counts:        # lookup 1
    value = counts["a"]  # lookup 2
else:
    value = 0

counts = {"a": 1}
try:
    value = counts["a"]  # lookup 1, and done
except KeyError:
    value = 0
# or simply: counts.get("a", 0)

class ConfigError(Exception):
    """Domain errors get their own class, so callers can catch precisely."""


def parse_port(raw: str) -> int:
    """Parse a port number, or raise ConfigError with a useful message.

    >>> parse_port("8080")
    8080
    >>> try:
    ...     parse_port("nope")
    ... except ConfigError as err:
    ...     print(err)
    port must be an integer, got 'nope'
    """
    try:
        port = int(raw)
    except ValueError as err:
        # `from err` keeps the original traceback attached as __cause__.
        raise ConfigError(f"port must be an integer, got {raw!r}") from err

    if not 1 <= port <= 65535:
        raise ConfigError(f"port out of range: {port}")

    return port

The four clauses

Typing beyond the basics

from typing import Iterable, Protocol, TypedDict, TypeVar

T = TypeVar("T")


def first(items: Iterable[T]) -> T | None:
    """Generic: whatever type goes in is the type that comes out.

    >>> first([1, 2])
    1
    >>> first([]) is None
    True
    """
    for item in items:
        return item
    return None


class SupportsLessThan(Protocol):
    """Structural typing. Anything with __lt__ satisfies this, no inheritance."""

    def __lt__(self, other: object) -> bool: ...


class Row(TypedDict):
    """A dict with a KNOWN shape, so the type checker can catch typos."""

    name: str
    score: int


row: Row = {"name": "ann", "score": 10}

Concurrency and the GIL

from concurrent.futures import ThreadPoolExecutor


def fetch(url: str) -> int:
    """Stands in for an I/O call. The GIL is released while waiting."""
    return len(url)


with ThreadPoolExecutor(max_workers=4) as pool:
    lengths = list(pool.map(fetch, ["a", "bb", "ccc"]))     # [1, 2, 3]

Performance habits

import timeit

# Measure, do not guess.
elapsed = timeit.timeit("sum(range(100))", number=100)      # seconds for 100 runs


def hot_loop(values: list[int]) -> list[int]:
    """Bind the method once, outside the loop.

    Each `out.append(...)` inside a loop is an attribute lookup plus a call.
    Binding it to a local name removes the lookup from every iteration.

    >>> hot_loop([1, 2, 3])
    [2, 4, 6]
    """
    out: list[int] = []
    append = out.append          # O(1) once, instead of O(1) n times
    for value in values:
        append(value * 2)
    return out