第 22 章 函数式编程
22.1 Python 中的函数式风格
Python 不是纯函数式语言，但支持函数式编程风格：
# 命令式风格
result = []
for x in range(10):
if x % 2 == 0:
result.append(x ** 2)
# 函数式风格
result = list(map(lambda x: x ** 2, filter(lambda x: x % 2 == 0, range(10))))
# Pythonic 风格（推导式 — 通常更推荐）
result = [x ** 2 for x in range(10) if x % 2 == 0]函数式编程的核心思想：
- 纯函数：相同输入总是返回相同输出，无副作用
- 不可变数据：不修改数据，而是创建新数据
- 函数是一等公民：函数可以作为参数、返回值
- 高阶函数：接收或返回函数的函数
- 惰性求值：按需计算
22.2 纯函数与不可变数据
纯函数
# 纯函数 — 没有副作用，结果只取决于输入
def add(a, b):
return a + b
def square_all(numbers):
return [x ** 2 for x in numbers]
# 不纯的函数 — 有副作用
total = 0
def impure_add(x):
global total
total += x # 副作用：修改全局状态
return total不可变数据
# 使用元组代替列表
point = (3, 4)
# point[0] = 5 # TypeError
# frozenset 代替 set
s = frozenset([1, 2, 3])
# 不修改原始数据，返回新数据
def add_element(lst, element):
return [*lst, element] # 创建新列表
original = [1, 2, 3]
new_list = add_element(original, 4)
print(original) # [1, 2, 3] — 不变
print(new_list) # [1, 2, 3, 4]
# frozen dataclass
from dataclasses import dataclass
@dataclass(frozen=True)
class Point:
x: float
y: float
def move(self, dx, dy):
return Point(self.x + dx, self.y + dy) # 返回新对象
p1 = Point(0, 0)
p2 = p1.move(3, 4) # p1 不变，p2 是新对象22.3 高阶函数深入
# 函数作为参数
def apply_twice(func, value):
return func(func(value))
print(apply_twice(lambda x: x + 3, 7)) # 13
print(apply_twice(lambda x: x ** 2, 3)) # 81
# 函数作为返回值
def make_power(n):
def power(x):
return x ** n
return power
square = make_power(2)
cube = make_power(3)
print(square(5)) # 25
print(cube(5)) # 125
# 函数工厂
def make_validator(min_val, max_val):
def validate(value):
return min_val <= value <= max_val
return validate
is_percentage = make_validator(0, 100)
is_adult_age = make_validator(18, 120)
print(is_percentage(50)) # True
print(is_adult_age(15)) # False22.4 functools 模块精讲
22.4.1 partial 偏函数
固定函数的部分参数，生成新函数：
from functools import partial
# 固定 base=2 的 int 函数
binary_to_int = partial(int, base=2)
print(binary_to_int("1010")) # 10
print(binary_to_int("1111")) # 15
# 固定 base=16
hex_to_int = partial(int, base=16)
print(hex_to_int("FF")) # 255
# 实用：日志函数
import logging
debug_log = partial(logging.log, logging.DEBUG)
error_log = partial(logging.log, logging.ERROR)
# 实用：数据库查询
def query(table, conditions=None, limit=None, offset=0):
print(f"SELECT * FROM {table} WHERE {conditions} LIMIT {limit} OFFSET {offset}")
query_users = partial(query, "users")
query_users(conditions="active=1", limit=10)
# partial 与 lambda 的区别
# partial 更高效（不创建新函数体），且能被 pickle 序列化22.4.2 lru_cache / cache 缓存
from functools import lru_cache, cache
# lru_cache — LRU 淘汰策略，限制缓存大小
@lru_cache(maxsize=128)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(100)) # 瞬间完成
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)
fibonacci.cache_clear() # 清除缓存
# cache — 无大小限制（Python 3.9+），等价于 lru_cache(maxsize=None)
@cache
def factorial(n):
return n * factorial(n - 1) if n else 1
# 缓存的注意事项：
# 1. 参数必须是可哈希的
# 2. 注意内存使用（无限缓存可能导致内存泄漏）
# 3. 不适合有副作用的函数22.4.3 singledispatch 单分派泛函数
根据第一个参数的类型分派到不同实现：
from functools import singledispatch
@singledispatch
def process(data):
"""默认处理"""
raise TypeError(f"不支持的类型: {type(data)}")
@process.register(str)
def _(data):
return f"字符串: {data.upper()}"
@process.register(int)
def _(data):
return f"整数: {data * 2}"
@process.register(list)
def _(data):
return f"列表({len(data)}个元素): {data}"
@process.register(dict)
def _(data):
return f"字典({len(data)}个键): {list(data.keys())}"
print(process("hello")) # 字符串: HELLO
print(process(42)) # 整数: 84
print(process([1, 2, 3])) # 列表(3个元素): [1, 2, 3]
print(process({"a": 1})) # 字典(1个键): ['a']# 也可以用类型注解注册（Python 3.7+）
@singledispatch
def serialize(obj):
raise TypeError(f"无法序列化 {type(obj)}")
@serialize.register
def _(obj: int) -> str:
return str(obj)
@serialize.register
def _(obj: float) -> str:
return f"{obj:.2f}"
@serialize.register
def _(obj: list) -> str:
return "[" + ", ".join(serialize(item) for item in obj) + "]"22.4.4 reduce 折叠操作
from functools import reduce
# 累积求和
total = reduce(lambda acc, x: acc + x, [1, 2, 3, 4, 5]) # 15
# 等价过程: ((((1+2)+3)+4)+5)
# 求最大值
max_val = reduce(lambda a, b: a if a > b else b, [3, 1, 4, 1, 5, 9]) # 9
# 展平嵌套列表
nested = [[1, 2], [3, 4], [5, 6]]
flat = reduce(lambda a, b: a + b, nested) # [1, 2, 3, 4, 5, 6]
# 带初始值
product = reduce(lambda acc, x: acc * x, [1, 2, 3, 4, 5], 1) # 120
# 字典合并
dicts = [{"a": 1}, {"b": 2}, {"c": 3}]
merged = reduce(lambda a, b: {**a, **b}, dicts)
# {'a': 1, 'b': 2, 'c': 3}
# 管道执行
def pipe(value, *functions):
return reduce(lambda v, f: f(v), functions, value)
result = pipe(
" Hello, World! ",
str.strip,
str.lower,
lambda s: s.replace(",", ""),
str.split,
)
print(result) # ['hello', 'world!']22.5 operator 模块
operator 模块提供运算符的函数版本，比 lambda 更高效、更可读：
import operator
# 算术运算
operator.add(3, 4) # 7 — 替代 lambda a, b: a + b
operator.mul(3, 4) # 12
operator.pow(2, 10) # 1024
# 比较运算
operator.lt(3, 5) # True
operator.eq("a", "a") # True
# 属性和元素获取
from operator import itemgetter, attrgetter, methodcaller
# itemgetter — 获取字典/序列的元素
students = [
{"name": "Alice", "age": 30, "score": 90},
{"name": "Bob", "age": 25, "score": 85},
{"name": "Charlie", "age": 35, "score": 92},
]
# 按 score 排序
sorted(students, key=itemgetter("score"))
# 多级排序
sorted(students, key=itemgetter("score", "name"))
# 提取多个字段
get_name_score = itemgetter("name", "score")
print(get_name_score(students[0])) # ('Alice', 90)
# attrgetter — 获取对象属性
from collections import namedtuple
Point = namedtuple("Point", ["x", "y"])
points = [Point(3, 4), Point(1, 2), Point(5, 0)]
sorted(points, key=attrgetter("x")) # 按 x 排序
sorted(points, key=attrgetter("y", "x")) # 先按 y，再按 x
# methodcaller — 调用方法
names = [" Alice ", " Bob ", " Charlie "]
list(map(methodcaller("strip"), names))
# ['Alice', 'Bob', 'Charlie']
# 带参数的方法调用
list(map(methodcaller("replace", " ", "_"), ["hello world", "foo bar"]))
# ['hello_world', 'foo_bar']22.6 函数组合与管道模式
函数组合
def compose(*functions):
"""从右到左组合函数: compose(f, g, h)(x) == f(g(h(x)))"""
def composed(x):
result = x
for f in reversed(functions):
result = f(result)
return result
return composed
# 数据处理管道
clean = compose(
str.strip,
str.lower,
lambda s: s.replace(" ", " "),
)
print(clean(" Hello World ")) # "hello world"管道模式
def pipe(value, *functions):
"""从左到右管道式处理: pipe(x, f, g, h) == h(g(f(x)))"""
for f in functions:
value = f(value)
return value
# 数据处理
result = pipe(
[1, -2, 3, -4, 5, -6],
lambda lst: [abs(x) for x in lst], # 取绝对值
lambda lst: [x for x in lst if x > 2], # 过滤
sorted, # 排序
lambda lst: lst[:3], # 取前 3
)
print(result) # [3, 4, 5]函数式工具集
from functools import reduce, partial
from operator import add, mul
# 组合使用
def sum_of_squares(numbers):
return reduce(add, map(lambda x: x**2, numbers))
# 更函数式的版本
sum_of_squares = compose(
partial(reduce, add),
partial(map, lambda x: x**2),
)
# 函数式错误处理（类似 Maybe/Option 模式）
def safe_divide(a, b):
return a / b if b != 0 else None
def safe_sqrt(x):
return x ** 0.5 if x is not None and x >= 0 else None
def safe_pipe(value, *functions):
for f in functions:
if value is None:
return None
value = f(value)
return value
result = safe_pipe(16, safe_sqrt, lambda x: x + 1, lambda x: safe_divide(x, 0))
print(result) # None（除以零返回 None，后续不再执行）本章小结：Python 的函数式编程不追求纯粹，而是实用主义——在合适的场景使用函数式风格可以让代码更简洁。
functools模块是核心工具箱：partial简化参数、lru_cache缓存计算、singledispatch实现类型分派、reduce累积折叠。operator模块提供的itemgetter/attrgetter在排序和数据提取中极为实用。