Chapter 25: Functional Programming
You've been writing Python in an imperative style — telling the computer how to do things step by step. Functional programming is a different approach: you describe what you want using functions that take inputs and return outputs, with no side effects and no shared state.
Python is not a purely functional language. But it borrows the best ideas from functional programming, and knowing them makes your code shorter, easier to test, and easier to reason about.
This chapter covers pure functions, immutability, map, filter, reduce, functools, and how to think functionally in a Python context.
Pure Functions
A pure function always returns the same output for the same input, and causes no side effects. No printing, no modifying globals, no writing to files — just input in, output out.
# Pure — same input always gives same output
def add(a, b):
return a + b
def square(n):
return n ** 2
def to_celsius(fahrenheit):
return (fahrenheit - 32) * 5 / 9
# Impure — depends on or modifies external state
total = 0
def add_to_total(n): # modifies global state
global total
total += n
def get_timestamp(): # result depends on the clock, not just inputs
import time
return time.time()
def save_user(user): # side effect: writes to disk
with open("users.txt", "a") as f:
f.write(str(user) + "\n")
Pure functions are easy to test (just call them with inputs, check outputs), easy to reason about (they don't depend on anything outside themselves), and safe to run in parallel (no shared state to corrupt).
You can't write a whole program with only pure functions — at some point you need to read input, write output, interact with the world. The functional style says: push the impure parts to the edges, keep the core pure.
# Functional core, imperative shell
# ─────────────────────────────────
# Pure core — no side effects
def parse_line(line):
parts = line.strip().split(",")
return {"name": parts[0], "score": float(parts[1])}
def is_passing(record):
return record["score"] >= 60
def format_record(record):
status = "PASS" if is_passing(record) else "FAIL"
return f"{record['name']:<15} {record['score']:>6.1f} {status}"
def compute_average(records):
if not records:
return 0.0
return sum(r["score"] for r in records) / len(records)
# Impure shell — reads/writes files, talks to the world
def process_scores(input_path, output_path):
with open(input_path) as f:
lines = f.readlines()
records = [parse_line(line) for line in lines if line.strip()]
passing = [r for r in records if is_passing(r)]
avg = compute_average(records)
with open(output_path, "w") as f:
for r in records:
f.write(format_record(r) + "\n")
f.write(f"\nAverage: {avg:.1f}\n")
f.write(f"Passing: {len(passing)}/{len(records)}\n")
return avg, len(passing)
The pure functions (parse_line, is_passing, format_record, compute_average) are trivial to unit-test. The impure shell (process_scores) is easy to read because it just calls the pure functions in order.
Immutability
Functional programming favors immutable data — data that doesn't change after creation. Instead of modifying existing data, you create new data.
# Mutable style — modify in place
def add_tax_mutable(items, rate):
for item in items:
item["price"] *= (1 + rate) # modifies the original dicts
return items
# Immutable style — create new data, leave original unchanged
def add_tax(items, rate):
return [
{**item, "price": item["price"] * (1 + rate)}
for item in items
]
cart = [
{"name": "Book", "price": 20.00},
{"name": "Pen", "price": 2.50},
{"name": "Laptop", "price": 999.99},
]
cart_with_tax = add_tax(cart, 0.08)
print(cart[0]["price"]) # 20.00 — original unchanged
print(cart_with_tax[0]["price"]) # 21.60 — new data
Immutability prevents a whole class of bugs where one part of your code modifies data that another part expects to stay the same. It makes functions predictable and programs easier to debug.
Python's built-in immutable types — tuple, frozenset, str — enforce this at the language level. For custom classes, @dataclass(frozen=True) (from Chapter 19) gives you immutable objects.
map — Transform Every Element
map(func, iterable) applies a function to every item in an iterable and returns a lazy iterator of results.
numbers = [1, 2, 3, 4, 5]
# Imperative
squared = []
for n in numbers:
squared.append(n ** 2)
# map
squared = list(map(lambda n: n ** 2, numbers))
# List comprehension (often preferred in Python)
squared = [n ** 2 for n in numbers]
All three do the same thing. map is lazy — it doesn't compute anything until you iterate:
result = map(str, range(5)) # nothing computed yet
print(result) # <map object at 0x...>
print(list(result)) # ['0', '1', '2', '3', '4']
map with named functions reads very naturally:
def celsius_to_fahrenheit(c):
return c * 9 / 5 + 32
temps_c = [0, 20, 37, 100]
temps_f = list(map(celsius_to_fahrenheit, temps_c))
print(temps_f) # [32.0, 68.0, 98.6, 212.0]
map also accepts multiple iterables — the function must accept that many arguments:
a = [1, 2, 3]
b = [10, 20, 30]
sums = list(map(lambda x, y: x + y, a, b))
print(sums) # [11, 22, 33]
# Equivalent with zip
sums = [x + y for x, y in zip(a, b)]
filter — Keep Only Matching Elements
filter(func, iterable) returns a lazy iterator of items for which func(item) is truthy.
numbers = range(-5, 6)
# Keep only positives
positives = list(filter(lambda n: n > 0, numbers))
print(positives) # [1, 2, 3, 4, 5]
# Equivalent comprehension
positives = [n for n in numbers if n > 0]
words = ["apple", "banana", "cherry", "date", "elderberry", "fig"]
long_words = list(filter(lambda w: len(w) > 5, words))
print(long_words) # ['banana', 'cherry', 'elderberry']
Pass None as the function to filter out falsy values:
data = [0, 1, "", "hello", None, [], [1, 2], False, True]
truthy = list(filter(None, data))
print(truthy) # [1, 'hello', [1, 2], True]
reduce — Fold a Sequence into One Value
reduce(func, iterable) repeatedly applies func to pairs of elements until one value remains. It's in functools:
from functools import reduce
numbers = [1, 2, 3, 4, 5]
total = reduce(lambda acc, x: acc + x, numbers) # ((((1+2)+3)+4)+5) = 15
product = reduce(lambda acc, x: acc * x, numbers) # ((((1*2)*3)*4)*5) = 120
maximum = reduce(lambda acc, x: acc if acc > x else x, numbers) # 5
print(total, product, maximum) # 15 120 5
You can provide an initial value as the third argument:
# Flatten a list of lists
nested = [[1, 2], [3, 4], [5, 6]]
flat = reduce(lambda acc, lst: acc + lst, nested, [])
print(flat) # [1, 2, 3, 4, 5, 6]
# Count words in a list of sentences
sentences = ["hello world", "foo bar baz", "python"]
word_count = reduce(lambda acc, s: acc + len(s.split()), sentences, 0)
print(word_count) # 6
In Python, reduce is less common than explicit loops or comprehensions because readability matters more here. Use it when the fold pattern is genuinely the clearest expression of the idea.
functools — Your Functional Toolkit
functools has several powerful tools for functional-style programming.
partial — Fix some arguments
functools.partial creates a new function with some arguments pre-filled:
from functools import partial
def power(base, exponent):
return base ** exponent
square = partial(power, exponent=2)
cube = partial(power, exponent=3)
two_to = partial(power, base=2)
print(square(5)) # 25
print(cube(3)) # 27
print(two_to(10)) # 1024
# Useful with map — apply a function with a fixed extra argument
from functools import partial
def clamp(value, low, high):
return max(low, min(high, value))
clamp_score = partial(clamp, low=0, high=100)
raw_scores = [85, -5, 110, 72, 0, 100, 150]
clamped = list(map(clamp_score, raw_scores))
print(clamped) # [85, 0, 100, 72, 0, 100, 100]
lru_cache — Memoize with a size limit
from functools import lru_cache
@lru_cache(maxsize=128)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(100)) # 354224848179261915075
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)
@lru_cache is thread-safe, respects maxsize, and exposes cache_info() and cache_clear(). Use it whenever a pure function is called repeatedly with the same arguments.
cache — Unbounded memoization (Python 3.9+)
from functools import cache
@cache
def expensive(n):
return sum(range(n))
# Same as @lru_cache(maxsize=None) but slightly faster
total_ordering — Fill in comparison methods
from functools import total_ordering
@total_ordering
class Temperature:
def __init__(self, celsius):
self.celsius = celsius
def __eq__(self, other):
return self.celsius == other.celsius
def __lt__(self, other):
return self.celsius < other.celsius
# total_ordering fills in: __le__, __gt__, __ge__
temps = [Temperature(100), Temperature(0), Temperature(37)]
print(sorted(temps)[0].celsius) # 0
print(max(temps).celsius) # 100
print(Temperature(20) <= Temperature(37)) # True
reduce — Already covered above
singledispatch — Function overloading by type
from functools import singledispatch
@singledispatch
def process(value):
raise TypeError(f"Cannot process type {type(value).__name__}")
@process.register(int)
def _(value):
return f"Integer: {value * 2}"
@process.register(str)
def _(value):
return f"String: {value.upper()}"
@process.register(list)
def _(value):
return f"List of {len(value)} items"
@process.register(float)
def _(value):
return f"Float: {value:.2f}"
print(process(42)) # Integer: 84
print(process("hello")) # String: HELLO
print(process([1, 2, 3])) # List of 3 items
print(process(3.14)) # Float: 3.14
Lambda Functions
A lambda is an anonymous function defined in a single expression. It's useful for short, one-off functions passed as arguments.
# Named function
def double(n):
return n * 2
# Lambda equivalent
double = lambda n: n * 2
# In practice, lambdas shine as inline arguments
numbers = [3, 1, 4, 1, 5, 9, 2, 6]
print(sorted(numbers, key=lambda n: -n)) # [9, 6, 5, 4, 3, 2, 1, 1]
people = [("Alice", 30), ("Bob", 25), ("Carlos", 35)]
print(sorted(people, key=lambda p: p[1])) # sorted by age
When to use lambdas vs named functions:
- Use a lambda when the function is very short, used once, and its meaning is obvious from context.
- Use a named function when the function is more than one expression, used multiple times, or needs a docstring.
# Good lambda use — short, obvious
evens = list(filter(lambda n: n % 2 == 0, range(10)))
# Bad lambda use — hard to read, should be a named function
result = sorted(data, key=lambda x: (x["dept"].lower(), -x["salary"], x["name"]))
# Better
def sort_key(employee):
return (employee["dept"].lower(), -employee["salary"], employee["name"])
result = sorted(data, key=sort_key)
Function Composition
Composition means combining functions so the output of one feeds into the input of the next.
def compose(*funcs):
"""Return a function that applies funcs right-to-left."""
from functools import reduce
def composed(value):
return reduce(lambda v, f: f(v), reversed(funcs), value)
return composed
def pipe(*funcs):
"""Return a function that applies funcs left-to-right."""
from functools import reduce
def piped(value):
return reduce(lambda v, f: f(v), funcs, value)
return piped
import math
normalize = lambda x: x / 100
to_radians = math.radians
sine = math.sin
to_percent = lambda x: x * 100
# pipe: normalize -> radians -> sin -> percent
process = pipe(normalize, to_radians, sine, to_percent)
print(f"{process(9000):.2f}%") # 0.00% (sin of 90 degrees = 1.0 -> 100%)
process2 = pipe(normalize, to_radians, sine, to_percent)
print(f"{process2(4500):.2f}%") # 70.71% (sin 45° ≈ 0.707)
# Text processing pipeline
strip = str.strip
lower = str.lower
capitalize = str.capitalize
def remove_extra_spaces(text):
return " ".join(text.split())
clean_title = pipe(strip, lower, remove_extra_spaces, capitalize)
titles = [
" PYTHON ZERO TO HERO ",
" the QUICK brown FOX ",
" FUNCTIONAL programming IN python ",
]
for title in titles:
print(clean_title(title))
Output:
Python zero to hero
The quick brown fox
Functional programming in python
Putting It All Together: A Data Processing Library
"""
dataflow.py — A small functional data processing library.
"""
from functools import reduce, partial
from typing import Callable, Iterable, Any
def pipeline(*funcs: Callable) -> Callable:
"""Compose functions left-to-right into a single callable."""
def run(value):
return reduce(lambda v, f: f(v), funcs, value)
return run
def pluck(field: str) -> Callable:
"""Return a function that extracts field from a dict."""
return lambda record: record[field]
def where(predicate: Callable) -> Callable:
"""Return a function that filters an iterable."""
return lambda items: (item for item in items if predicate(item))
def transform(func: Callable) -> Callable:
"""Return a function that maps func over an iterable."""
return lambda items: (func(item) for item in items)
def sort_by(key: Callable, reverse: bool = False) -> Callable:
"""Return a function that sorts an iterable."""
return lambda items: sorted(items, key=key, reverse=reverse)
def group_by(key: Callable) -> Callable:
"""Return a function that groups items into a dict by key(item)."""
def grouper(items):
groups = {}
for item in items:
k = key(item)
groups.setdefault(k, []).append(item)
return groups
return grouper
def aggregate(func: Callable) -> Callable:
"""Return a function that reduces an iterable to one value."""
return lambda items: func(list(items))
def to_list(items: Iterable) -> list:
return list(items)
# ── Demo ──────────────────────────────────────────────────────────────────────
employees = [
{"name": "Alice", "dept": "Eng", "salary": 95_000},
{"name": "Bob", "dept": "Eng", "salary": 82_000},
{"name": "Carlos", "dept": "HR", "salary": 68_000},
{"name": "Diana", "dept": "HR", "salary": 71_000},
{"name": "Eve", "dept": "Sales", "salary": 74_000},
{"name": "Frank", "dept": "Eng", "salary": 105_000},
{"name": "Grace", "dept": "Sales", "salary": 88_000},
]
# ── Query 1: Names of engineers earning over $90k ─────────────────────────────
high_earners = pipeline(
where(lambda e: e["dept"] == "Eng" and e["salary"] > 90_000),
transform(pluck("name")),
sort_by(str.lower),
to_list,
)
print("High-earning engineers:", high_earners(employees))
# ['Alice', 'Frank']
# ── Query 2: Average salary per department ────────────────────────────────────
avg_by_dept = pipeline(
group_by(pluck("dept")),
)
groups = avg_by_dept(employees)
for dept, members in sorted(groups.items()):
avg = sum(e["salary"] for e in members) / len(members)
print(f" {dept:<8} avg ${avg:,.0f}")
# Eng avg $94,000
# HR avg $69,500
# Sales avg $81,000
# ── Query 3: Top earner per department ───────────────────────────────────────
for dept, members in sorted(groups.items()):
top = max(members, key=pluck("salary"))
print(f" {dept:<8} top: {top['name']} (${top['salary']:,})")
# Eng top: Frank ($105,000)
# HR top: Diana ($71,000)
# Sales top: Grace ($88,000)
What You Learned in This Chapter
- Pure functions always return the same output for the same input and cause no side effects. They're easy to test and reason about.
- Immutability — creating new data instead of modifying existing data — prevents a class of bugs and makes code predictable.
map(func, iterable)— transform every element lazily.filter(func, iterable)— keep only elements wherefuncreturns truthy.filter(None, iterable)removes falsy values.reduce(func, iterable, initial)— fold a sequence into a single value.functools.partial— fix some arguments to create a specialized function.functools.lru_cache/functools.cache— memoize pure functions automatically.functools.total_ordering— define__eq__and__lt__, get all comparisons.functools.singledispatch— function overloading based on argument type.- Lambda functions — short anonymous functions for use as inline arguments.
- Function composition —
pipeandcomposecombine functions into processing pipelines. - The functional core, imperative shell pattern keeps side effects isolated at the edges.
What's Next?
Chapter 26 covers Advanced Python Types — typing module annotations, TypeVar, Generic, Protocol, NamedTuple, TypedDict, and how type hints make large codebases maintainable. If you've ever wondered what list[int] or Optional[str] means in someone else's code, you're about to find out.