Functional Programming in Python

2026-09-20 · 8 min

Python is not a functional language, but 3.10 through 3.12 added enough that a functional style is now a real option rather than a strained one. This is what that looks like in code, and where it stops working.

Everything below runs on 3.12. match and the | union syntax need 3.10+; the itertools.batched examples need 3.12.

$ python3 --version
Python 3.12.4

§ 01

Immutable values

A frozen dataclass is a product type with structural equality and hashing, generated from the field list.

from dataclasses import dataclass
 
@dataclass(frozen=True, slots=True)
class Point:
    x: float
    y: float
 
>>> Point(1, 2) == Point(1, 2)
True
>>> {Point(1, 2), Point(1, 2)}   # hashable, so this works
{Point(x=1, y=2)}

frozen actually enforces immutability at runtime, unlike a plain dataclass or a NamedTuple field reassignment attempt.

>>> p = Point(1, 2)
>>> p.x = 99
Traceback (most recent call last):
dataclasses.FrozenInstanceError: cannot assign to field 'x'

Updates go through replace, which returns a new instance and leaves the original untouched.

from dataclasses import replace
 
>>> p1 = Point(1, 2)
>>> p2 = replace(p1, x=99)
>>> p1, p2
(Point(x=1, y=2), Point(x=99, y=2))

Nested immutability needs nested frozen types, plus a container that will not let you mutate in place. tuple over list; MappingProxyType or a frozen dict-like over dict.

from types import MappingProxyType
 
@dataclass(frozen=True, slots=True)
class Config:
    hosts: tuple[str, ...]
    ports: MappingProxyType[str, int]
 
>>> c = Config(hosts=("a", "b"), ports=MappingProxyType({"http": 80}))
>>> c.ports["http"] = 8080
TypeError: 'mappingproxy' object does not support item assignment

Validation belongs in __post_init__. Because the instance is frozen, even __post_init__ has to go through object.__setattr__ to set a derived field.

@dataclass(frozen=True, slots=True)
class Range:
    low: int
    high: int
 
    def __post_init__(self):
        if self.low > self.high:
            raise ValueError(f"{self.low} > {self.high}")
        object.__setattr__(self, "span", self.high - self.low)

§ 02

Sum types

A closed sum, as a union of frozen dataclasses. There is no enum-with-payload in the language; this is the idiomatic substitute.

from dataclasses import dataclass
 
@dataclass(frozen=True, slots=True)
class Circle:
    r: float
 
@dataclass(frozen=True, slots=True)
class Rect:
    w: float
    h: float
 
@dataclass(frozen=True, slots=True)
class Tri:
    b: float
    h: float
 
type Shape = Circle | Rect | Tri   # PEP 695, 3.12

match with class patterns destructures each variant. This is the closest thing Python has to pattern matching over an algebraic type.

import math
 
def area(s: Shape) -> float:
    match s:
        case Circle(r=r):
            return math.pi * r * r
        case Rect(w=w, h=h):
            return w * h
        case Tri(b=b, h=h):
            return 0.5 * b * h

Positional patterns work when the dataclass declares __match_args__, which dataclasses generate automatically from the field order.

>>> Circle.__match_args__
('r',)
 
def area(s: Shape) -> float:
    match s:
        case Circle(r):
            return math.pi * r * r
        case Rect(w, h):
            return w * h
        case Tri(b, h):
            return 0.5 * b * h

There is no exhaustiveness check. A static checker can flag a missing case if the union is a type alias and every branch narrows it, but nothing stops you shipping code that omits one.

def area(s: Shape) -> float:
    match s:
        case Circle(r):
            return math.pi * r * r
        case Rect(w, h):
            return w * h
        # Tri unhandled: falls through, area() returns None

Guard against that by matching an explicit wildcard that raises, so a missing case fails loudly instead of returning None.

def area(s: Shape) -> float:
    match s:
        case Circle(r):
            return math.pi * r * r
        case Rect(w, h):
            return w * h
        case Tri(b, h):
            return 0.5 * b * h
        case _:
            raise AssertionError(f"unhandled: {s!r}")

Nested patterns, guards, and matching on sequences and mappings in the same statement.

def describe(event: object) -> str:
    match event:
        case {"type": "click", "x": x, "y": y} if x < 0 or y < 0:
            return "click out of bounds"
        case {"type": "click", "x": x, "y": y}:
            return f"click at ({x}, {y})"
        case {"type": "key", "code": ("ctrl", key)}:
            return f"ctrl+{key}"
        case [first, *rest] if rest:
            return f"batch starting with {first}, {len(rest)} more"
        case _:
            return "unknown"

§ 03

optional and result, as types

X | None is the idiomatic optional. mypy and pyright both narrow it after an is None / is not None check.

def find(xs: list[int], target: int) -> int | None:
    for i, x in enumerate(xs):
        if x == target:
            return i
    return None
 
def report(xs: list[int], target: int) -> str:
    i = find(xs, target)
    if i is None:
        return "not found"
    return f"found at {i}"   # i: int here, narrowed

A Result type, built the same way as the shape example above: a closed sum over two frozen dataclasses, generic in the payload and the error.

from dataclasses import dataclass
from typing import Generic, TypeVar
 
T = TypeVar("T")
E = TypeVar("E")
 
@dataclass(frozen=True, slots=True)
class Ok(Generic[T]):
    value: T
 
@dataclass(frozen=True, slots=True)
class Err(Generic[E]):
    error: E
 
type Result[T, E] = Ok[T] | Err[E]

Chaining without exceptions. Each step returns a Result, and match decides whether to continue or short-circuit.

def parse_int(s: str) -> Result[int, str]:
    try:
        return Ok(int(s))
    except ValueError:
        return Err(f"not an int: {s!r}")
 
def positive(n: int) -> Result[int, str]:
    return Ok(n) if n > 0 else Err(f"not positive: {n}")
 
def and_then[T, U, E](r: Result[T, E], f) -> Result[U, E]:
    match r:
        case Ok(value):
            return f(value)
        case Err(_):
            return r
 
>>> and_then(parse_int("42"), positive)
Ok(value=42)
>>> and_then(parse_int("-5"), positive)
Err(error='not positive: -5')
>>> and_then(parse_int("x"), positive)
Err(error="not an int: 'x'")

§ 04

Lazy pipelines with itertools

Generator expressions and itertools compose without building intermediate lists, the same way a range pipeline does elsewhere. Nothing runs until you iterate.

from itertools import islice
 
def squares_of_evens(xs):
    return (n * n for n in xs if n % 2 == 0)
 
>>> list(islice(squares_of_evens(range(1, 1000000)), 5))
[4, 16, 36, 64, 100]
# the other 999,995 elements were never touched

An infinite generator, made finite downstream. count() never terminates on its own.

from itertools import count, islice
 
def naturals():
    yield from count(0)
 
def fibs():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
 
>>> list(islice((n for n in fibs() if n % 2 == 0), 10))
[0, 2, 8, 34, 144, 610, 2584, 10946, 46368, 196418]

The itertools worth knowing by name, grouped by what they do.

chain, chain.from_iterable   concatenate iterables
groupby                      group consecutive equal keys
islice                       slice a lazily, no materialization
takewhile, dropwhile         stop, or start, on a predicate
tee                          split one iterator into independent copies
pairwise                     (x0,x1), (x1,x2), ...
batched                      fixed-size chunks, 3.12+
product, permutations, combinations   the combinatorial trio

groupby only groups consecutive runs, which is the thing people trip over. Sort by the key first if the input is not already grouped.

from itertools import groupby
 
words = ["ant", "bee", "bear", "cat", "cow", "ant"]
 
# wrong: "ant" appears in two different groups
>>> [(k, list(g)) for k, g in groupby(words, key=lambda w: w[0])]
[('a', ['ant']), ('b', ['bee', 'bear']), ('c', ['cat', 'cow']), ('a', ['ant'])]
 
# right: sort by the key first
>>> [(k, list(g)) for k, g in groupby(sorted(words), key=lambda w: w[0])]
[('a', ['ant', 'ant']), ('b', ['bear', 'bee']), ('c', ['cat', 'cow'])]

pairwise and batched, both 3.10+ and 3.12 respectively, replace loops that used to track an index by hand.

from itertools import pairwise, batched
 
>>> list(pairwise([1, 2, 3, 4]))
[(1, 2), (2, 3), (3, 4)]
 
>>> list(batched(range(10), 3))
[(0, 1, 2), (3, 4, 5), (6, 7, 8), (9,)]

A realistic pipeline: read lines lazily, parse, filter, group, all without holding the file in memory.

from itertools import groupby
from dataclasses import dataclass
 
@dataclass(frozen=True, slots=True)
class LogLine:
    level: str
    message: str
 
def parse_line(line: str) -> LogLine | None:
    parts = line.rstrip("\n").split(" ", 1)
    if len(parts) != 2:
        return None
    level, message = parts
    return LogLine(level, message)
 
def errors_by_run(path: str):
    with open(path) as f:
        lines = (parse_line(l) for l in f)
        parsed = (l for l in lines if l is not None)
        errors = (l for l in parsed if l.level == "ERROR")
        yield from groupby(errors, key=lambda l: l.message.split(":")[0])

§ 05

functools: reduce, partial, cache

reduce is fold_left. Python deliberately kept it out of builtins; Guido's own argument was that a named loop reads better for anything nontrivial, which is worth taking seriously rather than reaching for reduce by reflex.

from functools import reduce
import operator
 
>>> reduce(operator.add, [1, 2, 3, 4], 0)
10
>>> reduce(operator.mul, range(1, 6), 1)   # factorial
120
 
# the loop, which most reviewers will prefer for anything with a body:
total = 0
for x in [1, 2, 3, 4]:
    total += x

partial fixes leading arguments, and the result is picklable in a way a lambda closing over the same values is not, which matters the moment it needs to cross a multiprocessing.Pool boundary.

from functools import partial
 
def power(base: float, exponent: float) -> float:
    return base ** exponent
 
square = partial(power, exponent=2)
cube   = partial(power, exponent=3)
 
>>> square(5), cube(5)
(25, 125)
 
from multiprocessing import Pool
>>> with Pool() as p:
...     p.map(square, range(5))
[0, 1, 4, 9, 16]

cache and lru_cache memoize on argument identity. Arguments must be hashable, which is another reason frozen dataclasses and tuples earn their keep.

from functools import cache
 
@cache
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)
 
>>> fib(80)
23416728348467685
>>> fib.cache_info()
CacheInfo(hits=78, misses=81, maxsize=None, currsize=81)

reduce composed with a pipeline: fold over a lazily filtered, transformed range without ever materializing it.

from functools import reduce
import operator
 
>>> reduce(
...     operator.add,
...     (n * n for n in range(1, 1000000) if n % 2 == 0),
...     0,
... )
333332833333000000

§ 06

Composition and higher-order functions

Python has no built-in compose. Writing one is a few lines, and reduce is the natural tool for folding over an arbitrary number of functions.

from functools import reduce
from typing import Callable
 
def compose(*fns: Callable) -> Callable:
    def composed(x):
        return reduce(lambda acc, f: f(acc), reversed(fns), x)
    return composed
 
shout = compose(str.upper, lambda s: s + "!")
 
>>> shout("hello")
'HELLO!'

Pipe order, which reads left to right and is usually the more natural direction for a data pipeline.

def pipe(*fns: Callable) -> Callable:
    def piped(x):
        return reduce(lambda acc, f: f(acc), fns, x)
    return piped
 
process = pipe(str.strip, str.lower, lambda s: s.replace(" ", "_"))
 
>>> process("  Hello World  ")
'hello_world'

Decorators are function composition with syntax. A decorator that takes arguments is a function returning a function returning a function, which is worth writing out once to stop it feeling magic.

import time
from functools import wraps
 
def retry(times: int, delay: float = 0.1):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(times):
                try:
                    return fn(*args, **kwargs)
                except Exception:
                    if attempt == times - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator
 
@retry(times=3)
def flaky_call():
    ...

§ 07

Structural recursion, and where it breaks

A recursive walk over a tree, the way you would write it anywhere else.

@dataclass(frozen=True, slots=True)
class Leaf:
    value: int
 
@dataclass(frozen=True, slots=True)
class Node:
    left: "Tree"
    right: "Tree"
 
type Tree = Leaf | Node
 
def total(t: Tree) -> int:
    match t:
        case Leaf(value):
            return value
        case Node(left, right):
            return total(left) + total(right)

Python has no tail-call optimization, and CPython's default recursion limit is 1000. A deep or unbalanced tree, or a naive recursive fold over a long list, hits it fast.

>>> import sys; sys.getrecursionlimit()
1000
 
>>> def count_down(n): return 0 if n == 0 else count_down(n - 1)
>>> count_down(2000)
Traceback (most recent call last):
RecursionError: maximum recursion depth exceeded

Raising the limit moves the wall, it does not remove it; the real bound is the C stack, and past a point this segfaults the interpreter instead of raising.

import sys
sys.setrecursionlimit(100_000)
# fixes small overruns, still eventually crashes the process

The fix is the same one every non-tail-call language needs: rewrite as an explicit loop with your own stack.

def total_iterative(t: Tree) -> int:
    stack = [t]
    acc = 0
    while stack:
        node = stack.pop()
        match node:
            case Leaf(value):
                acc += value
            case Node(left, right):
                stack.append(left)
                stack.append(right)
    return acc

§ 08

Pattern matching a real payload

A worked example that puts most of the above together: decoding a JSON-shaped command, structurally, with no isinstance chain.

def handle(command: dict) -> Result[str, str]:
    match command:
        case {"op": "move", "dx": int(dx), "dy": int(dy)}:
            return Ok(f"move by ({dx}, {dy})")
        case {"op": "rotate", "degrees": (int() | float()) as deg}:
            return Ok(f"rotate {deg} degrees")
        case {"op": "batch", "commands": [*cmds]} if cmds:
            results = [handle(c) for c in cmds]
            if all(isinstance(r, Ok) for r in results):
                return Ok(f"batch of {len(cmds)} ok")
            return Err("batch had a failing command")
        case {"op": str(op)}:
            return Err(f"unknown op: {op}")
        case _:
            return Err("malformed command")
 
>>> handle({"op": "move", "dx": 1, "dy": -1})
Ok(value='move by (1, -1)')
>>> handle({"op": "batch", "commands": [{"op": "move", "dx": 1, "dy": 1}]})
Ok(value='batch of 1 ok')
>>> handle({"op": "teleport"})
Err(error='unknown op: teleport')

← older

Typeclasses in Depth

2026-09-19