Part 2 · Intermediate Python · Module 18

Functional Python

Functional programming is a style built from small, pure functions combined into bigger ones, with data that does not change underneath you. Python is not a purely functional language, but its functional tools (functools, operator, comprehensions, immutable types) make code easier to test, cache, parallelize, and reason about.

⏱ ~2 hoursPrerequisites: M14 Iterators & Generators, M15 Decorators

Pure functions and immutability

A function is pure when two things are true:

  1. Its result depends only on its arguments: the same inputs always give the same output. It does not read global variables, the clock, a random generator, files, or the network.
  2. It has no side effects: it does not modify its arguments or any global state, print, write files, or send messages. Its only effect is the value it returns.

A pure function call can be replaced by its result without changing the program, a property called referential transparency. That makes pure functions trivially testable (no setup, no mocks), safely cacheable (the same input always gives the same answer), and safe to run in parallel (nothing is shared).

# impure: mutates its argument, depends on global state
TAX = 0.2
def add_tax(prices):
    for i, p in enumerate(prices):
        prices[i] = p * (1 + TAX)
    return prices
# pure: new list out, nothing touched
def add_tax(prices, rate):
    return [p * (1 + rate) for p in prices]

Real programs must have side effects; a program that prints nothing and saves nothing is useless. The functional approach is to push effects to the edges. This architecture is often called functional core, imperative shell: a thin outer layer reads input, calls pure functions that hold all the logic, and writes the output.

flowchart LR
  A[read CSV file] -->|rows| B(parse)
  B --> C(validate)
  C --> D(enrich)
  D --> E(aggregate)
  E -->|summary| F[print / save]
  subgraph core [pure functional core]
    B
    C
    D
    E
  end
    

Remember: pure = output depends only on input, and no side effects. Keep logic pure; keep I/O at the edges.

map, filter, reduce vs comprehensions

Three classic higher-order functions process collections:

map and filter return lazy iterators (M14): nothing is computed until you loop over them, and they can be consumed only once.

In Python, comprehensions are usually clearer than map/filter with a lambda, and dedicated built-ins (sum, min, max, any, all, "".join, math.prod) beat reduce for common folds. Use map when you already have a named function (map(str.strip, lines), map(int, fields)), and reduce for folds with no built-in, such as merging a list of dicts or composing functions.

Remember: comprehensions for most transforms; map with named functions; reduce only when no built-in fold exists.

Higher-order functions and composition

You met higher-order functions in M15: functions that take or return functions. A particularly useful one is composition: combining f and g into a new function h(x) = f(g(x)). With composition you build a data pipeline as a list of small, individually tested steps.

Step through pipe. Notice that pipe itself does no work: it returns a closure, and the steps run only when that closure is called.

Remember: composition turns small functions into bigger ones; pipe lists steps in data-flow order.

functools.partial and the 3.14 Placeholder

Partial application means fixing some of a function's arguments now and supplying the rest later. functools.partial(fn, *args, **kwargs) returns a new callable that remembers those arguments. It is a cleaner, inspectable alternative to a lambda that only forwards arguments:

from functools import partial

parse_binary = partial(int, base=2)          # instead of: lambda s: int(s, base=2)
parse_binary("1011")                         # 11
print_err = partial(print, file=sys.stderr)  # instead of: lambda *a: print(*a, file=sys.stderr)

Until 3.14, partial could only fix arguments from the left. To pre-fill the second positional argument of a function that does not accept it by keyword (many built-ins, such as str.replace or pow), you needed a lambda.

Remember: partial pre-fills arguments; Placeholder lets you skip a positional slot so it is supplied later.

Caching: cache, lru_cache, cached_property

Because a pure function always returns the same output for the same input, you can store results and skip the work next time. This is memoization, and functools provides it as decorators:

The classic demonstration is the naive recursive Fibonacci, which recomputes the same subproblems exponentially many times. Toggle memoization in the call tree:

cached_property

@functools.cached_property computes an attribute the first time it is read and stores the result on the instance, so later reads are plain attribute lookups. It suits expensive derived values of objects that do not change. (It needs an instance __dict__, so it does not work with slots=True dataclasses.)

Remember: cache only pure functions with hashable arguments; check effectiveness with cache_info(); cached_property for expensive values of effectively immutable objects.

singledispatch and total_ordering

Dispatching on type

Sometimes one operation needs a different implementation per type: serializing a value, pretty-printing a node, computing the size of a shape. A long if isinstance(...) elif isinstance(...) chain must be edited every time a type is added. functools.singledispatch turns a function into a generic function: you register an implementation per type, and the call dispatches on the type of the first argument, following the class hierarchy (a bool uses the int implementation unless bool has its own).

For methods, use functools.singledispatchmethod, which dispatches on the first argument after self.

total_ordering recap

From M11: if a class defines __eq__ and one of __lt__, __le__, __gt__, __ge__, the class decorator @functools.total_ordering fills in the other three. It is another example of a decorator that adds behavior to a class (M15).

from functools import total_ordering

@total_ordering
class Version:
    def __init__(self, major: int, minor: int) -> None:
        self.key = (major, minor)
    def __eq__(self, other: object) -> bool:
        return isinstance(other, Version) and self.key == other.key
    def __lt__(self, other: Version) -> bool:   # no quotes needed in 3.14
        return self.key < other.key

Version(3, 14) >= Version(3, 12)   # True: __ge__ was generated

The operator module

Many lambdas just call an operator or fetch something: lambda a, b: a + b, lambda row: row["price"], lambda p: p.name. The operator module provides these as ready-made, faster, picklable, self-describing functions:

Instead ofUse
lambda a, b: a + boperator.add (also mul, sub, truediv, neg, not_…)
lambda r: r["price"]itemgetter("price")
lambda r: (r[2], r[0])itemgetter(2, 0) (returns a tuple)
lambda p: p.address.cityattrgetter("address.city") (dotted paths work)
lambda s: s.split(",")methodcaller("split", ",")

Remember: itemgetter, attrgetter, and methodcaller are the idiomatic key= functions; operator.add and friends replace arithmetic lambdas.

Immutable data

Pure functions are much easier to write when the data itself cannot change. An immutable object cannot be modified after creation; "changing" it means building a new object. That rules out a whole class of bugs where two parts of the program share an object and one of them mutates it (the aliasing problems from M02 and M06). Immutable objects are also hashable, so they can be dict keys, set members, and cache keys.

MutableImmutable counterpart"Update" by
listtuplet + (x,), slicing, unpacking
setfrozensetfs | {x}, fs - {x}
@dataclass@dataclass(frozen=True)dataclasses.replace(obj, field=new)
dicttypes.MappingProxyType (read-only view); frozendict in 3.15{**d, "k": v} or d | {"k": v}

Recursion vs iteration

Functional languages express repetition with recursion: a function that calls itself on a smaller input until it reaches a base case. Recursion is natural for recursive data (trees, nested lists, JSON) and divide-and-conquer algorithms. Python supports it, but with two practical limits:

Rules of thumb: recurse over recursive structures whose depth is small (a file tree, a parsed expression); iterate over long sequences. Any recursion can be converted to iteration with an explicit stack (a list you push and pop yourself), which is how you handle deep trees safely. Part 4's M28 Recursion & Backtracking goes much deeper.

Practice

Write compose(*fns) so that compose(f, g, h)(x) == f(g(h(x))), and pipe(*fns) so that pipe(f, g, h)(x) == h(g(f(x))). With no functions, both return the identity function. Neither may call any function until the composed function is called.

Hint

A loop inside a closure works: for pipe, apply each function in order; compose is pipe with the functions reversed.

Show solution
def pipe(*fns):
    def run(value):
        for fn in fns:
            value = fn(value)
        return value
    return run

def compose(*fns):
    return pipe(*reversed(fns))

Implement my_reduce(fn, iterable, *initial) with the same semantics as functools.reduce: fold left to right; if an initial value is given, start from it; if not, start from the first item; an empty iterable with no initial value raises TypeError; an empty iterable with an initial value returns it. Accept any iterable, including generators. Then use it to write merge_dicts(dicts) that merges a list of dicts left to right (later keys win) without modifying the inputs.

Hint

Call iter(iterable). Without an initial value, take the first item with next(it) inside try/except StopIteration. For merge_dicts, fold with lambda acc, d: acc | d starting from {}.

Show solution
def my_reduce(fn, iterable, *initial):
    it = iter(iterable)
    if initial:
        acc = initial[0]
    else:
        try:
            acc = next(it)
        except StopIteration:
            raise TypeError("my_reduce() of empty iterable with no initial value") from None
    for item in it:
        acc = fn(acc, item)
    return acc

def merge_dicts(dicts):
    return my_reduce(lambda acc, d: acc | d, dicts, {})

Using functools.partial (no lambdas, no def), create:

  • from_hex: parses a hexadecimal string, so from_hex("ff") == 255;
  • cube: cube(3) == 27, built from pow with a Placeholder;
  • strip_stars: strip_stars("**hi**") == "hi", built from str.strip with a Placeholder;
  • csv_line: joins its arguments with commas and no newline when printed, built from print, so csv_line("a", "b") prints a,b.
Hint

int takes base as a keyword; print takes sep and end. str.strip(s, chars) takes the string first, so the string is the placeholder.

Show solution
from functools import Placeholder, partial

from_hex = partial(int, base=16)
cube = partial(pow, Placeholder, 3)
strip_stars = partial(str.strip, Placeholder, "*")
csv_line = partial(print, sep=",", end="")

A robot starts at the top-left of a grid of rows × cols cells and may move only right or down. Write count_paths(rows, cols) that returns the number of distinct paths to the bottom-right cell, using recursion (count_paths(r, c) = count_paths(r - 1, c) + count_paths(r, c - 1), with a grid of one row or one column having exactly 1 path). Decorate it with functools.lru_cache(maxsize=None) (or cache) so that count_paths(16, 16) is instant.

Hint

Base case first: if rows == 1 or cols == 1: return 1. Put @cache directly above the def.

Show solution
from functools import cache

@cache
def count_paths(rows, cols):
    if rows == 1 or cols == 1:
        return 1
    return count_paths(rows - 1, cols) + count_paths(rows, cols - 1)

Write a singledispatch function to_jsonable(value) that converts values into things json.dumps accepts:

  • str, int, float, bool, and None are returned unchanged;
  • list and tuple become lists of converted items;
  • dict becomes a dict with str(key) keys and converted values;
  • set and frozenset become sorted lists of converted items;
  • datetime.date becomes its ISO string;
  • anything else raises TypeError.
Hint

Make the base function raise TypeError. Register one implementation for several types by stacking @to_jsonable.register(str), @to_jsonable.register(int), … on the same function. type(None) is NoneType; bool is covered by int.

Show solution
from datetime import date
from functools import singledispatch

@singledispatch
def to_jsonable(value):
    raise TypeError(f"cannot convert {type(value).__name__} to JSON")

@to_jsonable.register(str)
@to_jsonable.register(int)
@to_jsonable.register(float)
@to_jsonable.register(type(None))
def _(value):
    return value

@to_jsonable.register(list)
@to_jsonable.register(tuple)
def _(value):
    return [to_jsonable(v) for v in value]

@to_jsonable.register
def _(value: dict):
    return {str(k): to_jsonable(v) for k, v in value.items()}

@to_jsonable.register(set)
@to_jsonable.register(frozenset)
def _(value):
    return sorted(to_jsonable(v) for v in value)

@to_jsonable.register
def _(value: date):
    return value.isoformat()

Define a frozen dataclass Item(name: str, price: float, qty: int). Write pure functions:

  • restock(items, name, amount): returns a new tuple in which the item called name has qty increased by amount (use dataclasses.replace); other items are the same objects; raise KeyError if no item has that name.
  • most_valuable(items, n): the names of the n items with the highest price * qty, highest first, ties broken by name ascending.
  • by_price: a key function for sorting by price, created with operator.attrgetter.
Hint

For ties broken ascending while values sort descending, use the key lambda i: (-i.price * i.qty, i.name). For restock, build the tuple with a comprehension and track whether the name was found.

Show solution
from dataclasses import dataclass, replace
from operator import attrgetter

@dataclass(frozen=True)
class Item:
    name: str
    price: float
    qty: int

def restock(items, name, amount):
    if not any(item.name == name for item in items):
        raise KeyError(name)
    return tuple(replace(item, qty=item.qty + amount) if item.name == name else item for item in items)

def most_valuable(items, n):
    ranked = sorted(items, key=lambda i: (-i.price * i.qty, i.name))
    return [item.name for item in ranked[:n]]

by_price = attrgetter("price")

Mini-project: a composable data pipeline

Mini-projectSales report from small pure functions

Turn a messy CSV of sales lines into a regional summary, using a functional core (pure, composable steps) and an imperative shell (one main() that reads the file and prints the report).

date,region,rep,sku,qty,unit_price
2026-10-01,North ,ada,SKU-1,3,19.99
2026-10-01,south,alan,SKU-2,,5.00
2026-10-02,NORTH,grace,SKU-1,1,19.99
2026-10-02,east,linus,SKU-3,2,abc
2026-10-03,south,alan,SKU-3,4,7.50
pixi init sales-pipeline && cd sales-pipeline
pixi add pytest
pixi run python -m sales_pipeline data/sales.csv

Spec

  • Steps are separate pure functions over iterables of immutable records (a frozen dataclass Sale): parse_rows, normalize (strip and lower-case regions), validate (drop rows with missing or non-numeric fields, collecting the reasons), enrich (add total = qty * unit_price with dataclasses.replace), aggregate (totals per region).
  • pipeline = pipe(parse_rows, normalize, validate, enrich, aggregate), using your pipe from exercise 1.
  • Use operator.attrgetter/itemgetter for every sort and grouping key, and functools.partial to configure steps (for example partial(top_reps, n=3)).
  • Only main() touches the file system or prints.
  • Each step has at least two unit tests that need no files and no mocks
  • No step mutates its input (a test proves it by freezing or comparing before/after)
  • validate returns the good records and a list of rejection reasons (row number + why)
  • The steps are lazy where possible (generators), so a 1 GB file streams through in constant memory
  • main() is the only impure function, and an integration test runs it on a tmp_path CSV with capsys
  • Swapping the order of two independent steps, or adding a new one, is a one-line change

Stretch: add a singledispatch render step that formats the summary as text, JSON, or Markdown depending on the type of a format object; memoize an expensive currency-conversion step with lru_cache and report its cache_info().

Quiz

Recap

Going further