Level 6 · Python Fluency: Advanced Python

The earlier levels covered the Python that almost every program needs. This last level is about the parts of the language that explain how the rest of it works, and the tools you reach for when a program has outgrown the simple approach. Generators come back as full pipelines that pass work along with yield from. Properties turn out to be one case of a general mechanism, the descriptor, which you can use yourself to validate many attributes at once. __slots__ trades a little flexibility for a lot of memory when you have millions of small objects. Metaprogramming lets a class take part in its own creation, which is how plugin registries are built. asyncio lets one program wait for many slow things at once. And a section on performance shows how to measure before you optimise, and why a better algorithm almost always beats a faster line of code.

A word of caution comes with this level. These are powerful tools, and the most common mistake with them is using them when a plain function or class would do. Every section therefore starts, as always, with a real problem and the simple version, and only brings in the advanced construct when the simple version has a real cost.

The problems come from all three other paths: walking the trees of the Algorithms path lazily, validating the settings of a Statistics distribution or a Machine Learning model, keeping a million Monte Carlo particles in memory, registering interchangeable models by name, and making a slow Algorithms solution fast.

Advanced iteration

An in-order walk of a binary search tree visits its values in sorted order. A first version collects them into a list, and each recursive call builds a list of its own that the caller then copies:

class Tree:
    def __init__(self, value, left=None, right=None):
        self.value, self.left, self.right = value, left, right

def in_order(t):
    if t is None:
        return []
    return in_order(t.left) + [t.value] + in_order(t.right)

root = Tree(8, Tree(3, Tree(1), Tree(6)), Tree(10, None, Tree(14)))
print(in_order(root))

[1, 3, 6, 8, 10, 14]

The list concatenation copies values again at every level of the tree, and the whole result has to exist before the caller can see the first value, even when it only wants the smallest three. A generator yields values as it finds them, and yield from hands over every value of another generator (here, the recursive call), which is the generator form of "do the left subtree, then this node, then the right":

from itertools import islice

class Tree:
    def __init__(self, value, left=None, right=None):
        self.value, self.left, self.right = value, left, right

def in_order(t):
    if t is not None:
        yield from in_order(t.left)
        yield t.value
        yield from in_order(t.right)

root = Tree(8, Tree(3, Tree(1), Tree(6)), Tree(10, None, Tree(14)))
print(list(in_order(root)))
print(list(islice(in_order(root), 3)))       # stops after three values
print(next(v for v in in_order(root) if v > 7))

[1, 3, 6, 8, 10, 14]
[1, 3, 6]
8

Nothing is copied, and a caller that stops early stops the walk too: taking three values visits only the left part of the tree. (Very deep trees still need a loop with an explicit stack, as the Algorithms path explains, because every level of yield from is a level of recursion.)

The same idea scales up into pipelines of iterators, where each stage is a small generator or itertools function and data flows through one item at a time. Here is a Machine Learning training loop's data source: read records from several files as if they were one stream, drop bad rows, and cut the stream into mini-batches of three. chain.from_iterable joins the files, batched (Python 3.12 and later) makes the batches and pairwise gives neighbouring items:

import io
from itertools import chain, batched, pairwise, takewhile

files = [io.StringIO("1.5\n2.0\nbad\n"), io.StringIO("3.5\n\n4.0\n5.5\n6.0\n")]

def parse(lines):
    for line in lines:
        try:
            yield float(line)
        except ValueError:
            pass

values = parse(chain.from_iterable(files))
for batch in batched(values, 3):
    print(batch)

print(list(pairwise([1, 4, 9, 16])))
print(list(takewhile(lambda x: x < 10, [1, 4, 9, 16, 2])))

(1.5, 2.0, 3.5)
(4.0, 5.5, 6.0)
[(1, 4), (4, 9), (9, 16)]
[1, 4, 9]

At no point does the program hold more than one batch, however large the files are. That is the property that lets the Statistics path summarise a stream of any length.

Common mistakes: writing yield in_order(t.left) instead of yield from, which yields the generator object itself rather than its values; consuming the same iterator in two stages of a pipeline; forgetting that takewhile stops at the first failing item rather than filtering (use a comprehension to filter); and turning a pipeline into a list at every step "to look at it", which defeats the point.

Practice: Walking a Slow Network Drive, The Deepest Dip in an Account, The Warmest Hour of Every Day

Descriptors

A gradient-descent model in the Machine Learning path has several settings that must all be positive: the learning rate, the number of epochs, the batch size. Level 4's @property can validate each one. A first version does exactly that:

class Trainer:
    def __init__(self, learning_rate, epochs):
        self.learning_rate = learning_rate
        self.epochs = epochs

    @property
    def learning_rate(self):
        return self._learning_rate
    @learning_rate.setter
    def learning_rate(self, value):
        if value <= 0:
            raise ValueError(f"learning_rate must be positive, got {value}")
        self._learning_rate = value

    @property
    def epochs(self):
        return self._epochs
    @epochs.setter
    def epochs(self, value):
        if value <= 0:
            raise ValueError(f"epochs must be positive, got {value}")
        self._epochs = value

t = Trainer(0.1, 20)
try:
    t.epochs = 0
except ValueError as e:
    print(e)

epochs must be positive, got 0

Two settings take twenty lines, and the two properties differ only in their names. A third setting would be a third copy, and a typo in one copy (self._epochs in the learning-rate setter) would be hard to spot. A descriptor is the mechanism underneath property: an object stored on the class that controls how an attribute is read and written on every instance. A class with __get__ and __set__ methods is a descriptor, and __set_name__ tells it the attribute name it was assigned to, so one class can describe "a positive attribute" for any number of names:

class Positive:
    def __set_name__(self, owner, name):
        self.name = name
        self.store = "_" + name
    def __get__(self, obj, objtype=None):
        if obj is None:                  # accessed on the class itself
            return self
        return getattr(obj, self.store)
    def __set__(self, obj, value):
        if value <= 0:
            raise ValueError(f"{self.name} must be positive, got {value}")
        setattr(obj, self.store, value)

class Trainer:
    learning_rate = Positive()
    epochs = Positive()
    batch_size = Positive()

    def __init__(self, learning_rate, epochs, batch_size=32):
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.batch_size = batch_size

class Gaussian:                          # the same rule, reused in Statistics
    std = Positive()
    def __init__(self, mean, std):
        self.mean, self.std = mean, std

t = Trainer(0.1, 20)
print(t.learning_rate, t.epochs, t.batch_size)
for make in (lambda: Trainer(0.1, 0), lambda: Gaussian(0, -1)):
    try:
        make()
    except ValueError as e:
        print(e)

0.1 20 32
epochs must be positive, got 0
std must be positive, got -1

When Python evaluates t.epochs = 0, it finds epochs on the class, sees that it has a __set__ method and calls Positive.__set__(descriptor, t, 0) instead of storing the value directly. Reading t.epochs calls __get__ in the same way.

You have been using descriptors all along. property is a descriptor class. So are ordinary functions: a function stored on a class has a __get__ method that turns obj.method into a bound method with self already filled in, which is where self comes from. Understanding this makes the rest of Python's object model much less mysterious, even if you only write a descriptor of your own once in a while.

Common mistakes: storing the value on the descriptor itself (self.value = value inside __set__), which shares one value between every instance of the class; storing it under the same name as the descriptor, which calls __set__ again forever; and writing a descriptor for a single attribute, where a property is simpler and clearer.

Practice: One Rule for Every Setting, Centimetres, Inches and Everything in Between, Statistics That Recompute Only When They Must

Slots

A Monte Carlo simulation from the Statistics path moves a large number of particles, each with a position and a velocity. A first version is an ordinary class, and it has a bug that Python does not report:

class Particle:
    def __init__(self, x, y, vx, vy):
        self.x, self.y, self.vx, self.vy = x, y, vx, vy
    def step(self, dt):
        self.x += self.vx * dt
        self.y += self.vy * dt

p = Particle(0.0, 0.0, 1.0, 2.0)
p.step(0.5)
p.vY = 0.0            # typo: meant to stop the particle
p.step(0.5)
print(p.x, p.y, sorted(vars(p)))

1.0 2.0 ['vY', 'vx', 'vy', 'x', 'y']

An ordinary object keeps its attributes in a per-object dictionary, __dict__, so any new name can be added at any time. That is flexible, but it means a misspelt attribute silently creates a new one (the particle did not stop), and each object carries the space for that dictionary. Declaring __slots__ lists the attributes the class will ever have. Python then stores them in fixed places inside the object, with no dictionary at all:

import tracemalloc

class Particle:
    __slots__ = ("x", "y", "vx", "vy")
    def __init__(self, x, y, vx, vy):
        self.x, self.y, self.vx, self.vy = x, y, vx, vy
    def step(self, dt):
        self.x += self.vx * dt
        self.y += self.vy * dt

p = Particle(0.0, 0.0, 1.0, 2.0)
try:
    p.vY = 0.0
except AttributeError:
    print("AttributeError:", type(p).__name__, "has no slot 'vY'")

class PlainParticle:
    def __init__(self, x, y, vx, vy):
        self.x, self.y, self.vx, self.vy = x, y, vx, vy

def bytes_for(cls, n=100_000):
    tracemalloc.start()
    objects = [cls(0.0, 0.0, 1.0, 1.0) for _ in range(n)]
    used, _ = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return used

print(bytes_for(Particle) < 0.8 * bytes_for(PlainParticle))

AttributeError: Particle has no slot 'vY'
True

The typo is now an error at the line where it happens, and the objects are smaller: on current versions of Python, about a third less memory per object for a class like this one (older versions saved even more), plus slightly faster attribute access. For a million particles that is tens of megabytes. Dataclasses can do the same with one argument, @dataclass(slots=True).

The costs are that you cannot add attributes that are not listed, and a subclass needs its own __slots__ to keep the benefit. That is why __slots__ is for classes with many instances and a fixed set of fields (particles, points, tree nodes, tokens), not for every class you write.

Common mistakes: writing __slots__ = ("x"), which is the string "x" rather than a one-item tuple (write ("x",)); expecting vars(obj) or obj.__dict__ to work on a slotted object; giving a slot and a class attribute the same name, which raises ValueError; and adding __slots__ to a class with ten instances, where the saving is nothing and the restriction is real.

Practice: A Million Particles in a Box

Metaprogramming

A Machine Learning experiment script lets the user choose a model by name on the command line: "majority", "threshold" and so on. A first version keeps a dictionary from names to classes, maintained by hand:

class Model:
    def predict(self, x):
        raise NotImplementedError

class Majority(Model):
    def predict(self, x):
        return 1

class Threshold(Model):
    def predict(self, x):
        return int(x >= 0.5)

class Nearest(Model):              # added later...
    def predict(self, x):
        return int(round(x))

MODELS = {"majority": Majority, "threshold": Threshold}   # ...but not added here
print(sorted(MODELS))
print("nearest" in MODELS)

['majority', 'threshold']
False

Every new model must be added in two places, and forgetting the second is invisible until someone tries to use the model. Metaprogramming means code that works on code: here, the class definitions register themselves. Python calls a base class's __init_subclass__ method whenever a subclass is defined, and keyword arguments in the class line are passed to it:

class Model:
    registry = {}

    def __init_subclass__(cls, *, name, **kwargs):
        super().__init_subclass__(**kwargs)
        if not callable(getattr(cls, "predict", None)):
            raise TypeError(f"{cls.__name__} must define predict")
        Model.registry[name] = cls

class Majority(Model, name="majority"):
    def predict(self, x):
        return 1

class Threshold(Model, name="threshold"):
    def predict(self, x):
        return int(x >= 0.5)

class Nearest(Model, name="nearest"):
    def predict(self, x):
        return int(round(x))

print(sorted(Model.registry))
model = Model.registry["threshold"]()
print(model.predict(0.7))

['majority', 'nearest', 'threshold']
1

Defining the class is registering it, so the two can no longer drift apart, and the check that predict exists runs when the class is defined rather than when it is first used.

A class decorator does the same job from outside. It is a function that receives the finished class and returns it (or a modified version). @dataclass from Level 4 is one: it reads the class's annotations and adds methods. A registry decorator is only a few lines, and it works for classes that share no base class:

SCALERS = {}

def register(name):
    def decorator(cls):
        SCALERS[name] = cls
        cls.name = name
        return cls
    return decorator

@register("minmax")
class MinMax:
    def fit(self, xs):
        self.lo, self.hi = min(xs), max(xs)
        return self
    def transform(self, x):
        return (x - self.lo) / (self.hi - self.lo)

scaler = SCALERS["minmax"]().fit([2, 4, 10])
print(scaler.name, scaler.transform(6))

minmax 0.5

register("minmax") returns the decorator, which is why this one has brackets while @dataclass needs none: it is a decorator with settings, one level of closure deeper, as Level 5 described.

The honest advice on metaprogramming is to use it rarely. A registry, automatic validation of subclasses and generating repetitive methods are good uses; clever tricks that change what ordinary-looking code does are not. The deepest tool, metaclasses (the class of a class), is almost never needed now that __init_subclass__ and class decorators exist.

Common mistakes: forgetting to call super().__init_subclass__(**kwargs), which breaks other base classes that also use the hook; forgetting to return cls from a class decorator, which replaces the class with None; and reaching for metaprogramming when a plain dictionary or function would do.

Practice: Undo for Any Settings Object, Event Records That Know Their Own Fields, Pipeline Steps That Put Themselves in Order

asyncio basics

A dashboard needs three pieces of data from three slow sources: a server, a sensor and a database, each of which takes a moment to answer. A first version asks them one after the other. Here asyncio.sleep stands in for waiting on the network, and fetch is a coroutine function, written with async def:

import asyncio, time

async def fetch(name, seconds):
    await asyncio.sleep(seconds)        # waiting, not computing
    return f"{name} ready"

async def main():
    start = time.perf_counter()
    results = []
    for name, s in [("server", 0.1), ("sensor", 0.2), ("database", 0.15)]:
        results.append(await fetch(name, s))
    elapsed = time.perf_counter() - start
    print(results)
    print("took at least 0.45 s:", elapsed > 0.4)

asyncio.run(main())

['server ready', 'sensor ready', 'database ready']
took at least 0.45 s: True

The program spends almost all of that time doing nothing, waiting for one source before it even asks the next. Calling a coroutine function does not run it; it creates a coroutine, a paused computation. await runs a coroutine and, whenever that coroutine has to wait, hands control back to the event loop, which can run something else in the meantime. To wait for several things at once, start them all and then wait for all of them. asyncio.gather does both:

import asyncio, time

async def fetch(name, seconds):
    await asyncio.sleep(seconds)
    print("finished", name)
    return f"{name} ready"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("server", 0.1), fetch("sensor", 0.2), fetch("database", 0.15))
    elapsed = time.perf_counter() - start
    print(results)
    print("took about 0.2 s, not 0.45:", elapsed < 0.4)

asyncio.run(main())

finished server
finished database
finished sensor
['server ready', 'sensor ready', 'database ready']
took about 0.2 s, not 0.45: True

The total time is now that of the slowest source. The sources finish in order of their delays, but gather returns the results in the order you asked for them. asyncio.create_task starts a coroutine running in the background of the current loop and gives you a task to await later, so you can do other work in between:

import asyncio

async def countdown(label, n):
    for i in range(n, 0, -1):
        print(label, i)
        await asyncio.sleep(0.01)
    return label

async def main():
    task = asyncio.create_task(countdown("background", 2))   # scheduled in the running loop
    await countdown("main", 2)
    print("done:", await task)

asyncio.run(main())

main 2
background 2
main 1
background 1
done: background

The two countdowns take turns, each giving way at its await. Notice that main prints first: create_task only schedules the task, and it gets its first turn when main reaches an await and gives way. This is concurrency without threads: one worker switching between tasks at the points where they would otherwise wait. It helps with waiting (network, timers, user input), not with computing. A loop that crunches numbers never reaches an await, so nothing else runs until it finishes.

Two notes about the browser. This site runs Python in the browser, where threads are not available, so on this site concurrency always means coroutines like these. And the examples above end with asyncio.run(main()), which is how an ordinary Python program starts an event loop; in the browser an event loop is already running (the browser's own), and asyncio.run cannot start a second one. On this site your coroutines are run inside that existing loop, so write async def functions and await inside them, and use asyncio.create_task or asyncio.gather to run several at once.

Common mistakes: calling a coroutine without await (fetch("server", 1) alone does nothing and Python warns that it "was never awaited"); using time.sleep inside a coroutine, which blocks the whole loop, instead of await asyncio.sleep; writing await outside an async def; and expecting asyncio to speed up a calculation, which it cannot.

Practice: Ask Every Source at Once, A Crawler That Keeps Only a Few Downloads Running, An Event Loop of Your Own

Performance and profiling

A program checks which of 2,000 incoming IDs appear in a list of 5,000 known IDs, and it is too slow. A first version is the direct translation of the question:

import random, timeit

rng = random.Random(1)
known = [rng.randrange(10**6) for _ in range(5000)]
queries = [rng.randrange(10**6) for _ in range(2000)]

def count_known():
    return sum(q in known for q in queries)

print(count_known())
print(f"{timeit.timeit(count_known, number=3) / 3:.3f} s per run")

On a typical laptop this prints something like:

9
0.035 s per run

Before changing anything, the first rule of performance work: measure. timeit.timeit(func, number=k) runs a function k times and returns the total seconds, using the most precise clock available; dividing by k gives the time per run. Never judge speed from a single run, and never guess where the time goes: programmers are famously bad at it. For a whole program, time each stage with time.perf_counter() or count how often the key function is called (the count_calls decorator from Level 5). On your own computer, the cProfile module reports the time spent in every function.

Here the measurement points straight at the cause. q in known scans the list, up to 5,000 comparisons for each of 2,000 queries: ten million steps. That is an algorithmic problem, and the fix is an algorithmic one from Level 1: a set answers in in constant time.

import random, timeit

rng = random.Random(1)
known = [rng.randrange(10**6) for _ in range(5000)]
queries = [rng.randrange(10**6) for _ in range(2000)]
known_set = set(known)

def with_list():
    return sum(q in known for q in queries)

def with_set():
    return sum(q in known_set for q in queries)

print(with_list() == with_set())
t_list = timeit.timeit(with_list, number=3)
t_set = timeit.timeit(with_set, number=3)
print("set version more than 20 times faster:", t_list > 20 * t_set)

True
set version more than 20 times faster: True

In practice the gap is several hundred times, and it grows with the data: double both lists and the list version takes four times as long, while the set version takes twice as long. No amount of polishing the list version could match that. Press Run to measure both at several sizes:

import timeit
sizes = [250, 500, 1000, 2000]
t_list, t_set = [], []
for n in sizes:
    known, queries = list(range(0, 2 * n, 2)), list(range(n))
    known_set = set(known)
    t_list.append(timeit.timeit(lambda: sum(q in known for q in queries), number=3))
    t_set.append(timeit.timeit(lambda: sum(q in known_set for q in queries), number=3))
plot(sizes, t_list, label="list", logx=True, logy=True, xlabel="n (items in each list)", ylabel="seconds")
plot(sizes, t_set, label="set")

On these log-log axes the list line rises twice as steeply as the set line: about n^2 steps against about n.

Constant-factor speed-ups keep the same algorithm and make each step cheaper. The main ones in Python are using built-ins and comprehensions (their loops run in C), avoiding repeated work inside a loop (compute len(xs) or a lookup once, outside it), and choosing the right structure (deque.popleft() instead of list.pop(0), str.join instead of +=). They typically give between 1.2 and 5 times, which is worth having once the algorithm is right:

import timeit

nums = list(range(100_000))

def loop_total():
    total = 0
    for x in nums:
        total += x
    return total

def builtin_total():
    return sum(nums)

def squares_loop():
    out = []
    for x in nums:
        out.append(x * x)
    return out

def squares_comp():
    return [x * x for x in nums]

for f in (loop_total, builtin_total, squares_loop, squares_comp):
    print(f"{f.__name__:14}{timeit.timeit(f, number=20) / 20 * 1000:6.2f} ms")

A typical result (your numbers will differ, and code in the browser runs a few times slower than on a desktop Python):

loop_total      1.36 ms
builtin_total   0.26 ms
squares_loop    1.97 ms
squares_comp    1.65 ms

sum on a list of numbers runs entirely in C and wins by about five times; the comprehension saves the repeated out.append lookup and call, a smaller gain. Not every rewrite helps: sum(x * x for x in nums) is usually no faster than the plain loop, because the generator still runs Python code for each item. The differences are real but modest, they vary between Python versions, and only measuring tells you which ones matter. So the order of work is: first make it correct (with tests, Level 5); then measure; then look for a better algorithm or data structure (a set or dictionary instead of a scan, sorting once and using bisect, a heap instead of repeated min, memoisation instead of repeated recursion); and only then, if a measured hot spot remains, tune constant factors there.

Common mistakes: optimising before measuring; timing a single run, or timing code that includes printing; micro-tuning a loop that is O(n²) when an O(n) algorithm exists; and making code harder to read for a speed-up too small to notice. A clear program that is fast enough is better than an obscure one that is slightly faster.

Practice: Counting Events on a Slow Disk, The Fastest Requests Without Sorting the Day, Matching Orders to Customers Without Asking Twice

Practice

These problems are required on this path at this level; the path adds one problem from each of the level's topics, and one for each construct the others do not use, picked for you and kept.

• Walking a Slow Network Drive
• The Warmest Hour of Every Day
• When the Fast Average Overtakes the Slow One
• One Rule for Every Setting
• A Million Particles in a Box
• Centimetres, Inches and Everything in Between

Checklist

• Write recursive generators with yield from, and build lazy pipelines with chain, batched, pairwise, islice and takewhile.
• Explain how a descriptor's __get__, __set__ and __set_name__ work, write one to validate many attributes, and say why property and methods are descriptors.
• Use __slots__ (or @dataclass(slots=True)) for classes with many instances, and know what it costs.
• Register subclasses automatically with __init_subclass__ or a class decorator, and know when not to.
• Write coroutines with async def and await, run them concurrently with gather and create_task, and explain why this helps waiting but not computing, and how the browser runs them.
• Measure with timeit and time.perf_counter, tell algorithmic from constant-factor improvements, and apply them in that order.