Python Functional Programming Guide 2026: Pure Functions, map and Pipelines

# Pure function: same inputs → same outputs, no side effects
def add(a: int, b: int) -> int:
    return a + b  # pure: no external state, no I/O

def get_tax(price: float, rate: float = 0.1) -> float:
    return price * rate  # pure

# Impure function (has side effects)
total = 0
def add_to_total(n: int) -> None:  # modifies external state
    global total
    total += n  # side effect!

# Impure: same call, different result
def get_current_time():
    import time
    return time.time()  # side effect: depends on external state

from typing import Callable, TypeVar

T = TypeVar("T")
U = TypeVar("U")

# Functions that take or return functions
def compose(*fns: Callable) -> Callable:
    from functools import reduce
    return reduce(lambda f, g: lambda x: f(g(x)), fns)

def pipe(*fns: Callable) -> Callable:
    return reduce(lambda f, g: lambda x: g(f(x)), fns)

# Usage
clean_name = compose(str.title, str.strip, str.lower)
print(clean_name("  ALICE SMITH  "))  # "Alice Smith"

# Partial application
from functools import partial

def multiply(a: int, b: int) -> int:
    return a * b

double = partial(multiply, 2)
triple = partial(multiply, 3)

numbers = [1, 2, 3, 4, 5]
print(list(map(double, numbers)))  # [2, 4, 6, 8, 10]

# Memoize expensive functions
from functools import cache

@cache
def fib(n: int) -> int:
    if n < 2: return n
    return fib(n-1) + fib(n-2)

from functools import reduce

users = [
    {"name": "Alice", "age": 30, "active": True, "score": 8.5},
    {"name": "Bob", "age": 25, "active": False, "score": 7.2},
    {"name": "Carol", "age": 35, "active": True, "score": 9.1},
    {"name": "Dave", "age": 28, "active": True, "score": 6.8},
]

# map: transform each element
names = list(map(lambda u: u["name"], users))

# filter: keep matching elements
active_users = list(filter(lambda u: u["active"], users))

# reduce: fold into single value
total_score = reduce(lambda acc, u: acc + u["score"], users, 0.0)
avg_score = total_score / len(users)

# Comprehensions (preferred over map/filter in Python)
names = [u["name"] for u in users]
active_users = [u for u in users if u["active"]]
total_score = sum(u["score"] for u in users)

# When to use map/filter vs comprehension:
# map/filter: when working with existing function references
# Comprehension: for new logic (more Pythonic, more readable)
squares = list(map(lambda x: x**2, range(10)))  # map OK
squares = [x**2 for x in range(10)]              # better

from dataclasses import dataclass, replace
from typing import NamedTuple

# Frozen dataclass — immutable
@dataclass(frozen=True)
class Point:
    x: float
    y: float

    def translate(self, dx: float, dy: float) -> "Point":
        return replace(self, x=self.x + dx, y=self.y + dy)  # new object!

p1 = Point(1.0, 2.0)
p2 = p1.translate(3.0, 4.0)  # p1 unchanged
# p1.x = 10  # raises FrozenInstanceError

# NamedTuple — immutable, fast
class User(NamedTuple):
    id: int
    name: str
    email: str

user = User(1, "Alice", "alice@example.com")
updated = user._replace(name="Alice Smith")  # new User

# Immutable dict update patterns
original = {"a": 1, "b": 2, "c": 3}
updated = {**original, "b": 99}  # {a:1, b:99, c:3} — original unchanged

# list transformations that return new lists (don't mutate)
original = [1, 2, 3, 4, 5]
doubled = [x * 2 for x in original]  # new list
filtered = [x for x in original if x > 2]  # new list
sorted_list = sorted(original)  # new list (sorted() vs .sort())

from itertools import islice, chain, filterfalse
from functools import reduce

# Data pipeline as function composition
def process_orders(orders):
    pipeline = [
        # Filter: only completed orders
        lambda orders: filter(lambda o: o["status"] == "completed", orders),
        # Transform: calculate revenue per order
        lambda orders: map(lambda o: {**o, "revenue": o["price"] * o["qty"]}, orders),
        # Sort: by revenue descending
        lambda orders: sorted(orders, key=lambda o: o["revenue"], reverse=True),
        # Limit: top 10
        lambda orders: islice(orders, 10),
    ]
    return reduce(lambda data, fn: fn(data), pipeline, orders)

top_orders = list(process_orders(all_orders))

# Using toolz library for cleaner pipelines
from toolz import pipe, curry, compose

@curry
def filter_status(status, orders):
    return filter(lambda o: o["status"] == status, orders)

@curry
def add_revenue(orders):
    return map(lambda o: {**o, "revenue": o["price"] * o["qty"]}, orders)

result = pipe(
    all_orders,
    filter_status("completed"),
    add_revenue,
    list
)