函数式编程

把函数当变量传来传去——高阶函数、闭包和装饰器让代码优雅度飙升。

函数式编程地图

高阶函数

"""
高阶函数：接受函数作为参数，或返回函数
"""
# === lambda 表达式 ===
square = lambda x: x ** 2
add = lambda x, y: x + y
print(square(5))   # 25
print(add(3, 4))   # 7
# === map: 映射 ===
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x ** 2, numbers))
print(squared)  # [1, 4, 9, 16, 25]
# 等价列表推导式（更 Pythonic）
squared = [x ** 2 for x in numbers]
# === filter: 过滤 ===
evens = list(filter(lambda x: x % 2 == 0, range(10)))
print(evens)  # [0, 2, 4, 6, 8]
# 等价
evens = [x for x in range(10) if x % 2 == 0]
# === reduce: 累积 ===
from functools import reduce
total = reduce(lambda acc, x: acc + x, numbers)
print(total)  # 15
# 求最大值
maximum = reduce(lambda a, b: a if a > b else b, numbers)
print(maximum)  # 5
# === sorted: 自定义排序 ===
words = ["banana", "apple", "cherry", "date"]
by_length = sorted(words, key=len)
print(by_length)  # ['date', 'apple', 'banana', 'cherry']
# 多条件排序
students = [("张三", 85), ("李四", 92), ("王五", 85), ("赵六", 78)]
by_score = sorted(students, key=lambda s: (-s[1], s[0]))
print(by_score)  # 分数降序，同分按名字升序

闭包

"""
闭包：函数 + 环境变量
"""
# === 基本闭包 ===
def make_multiplier(factor: int):
"""创建乘法器"""
def multiply(x: int) -> int:
return x * factor  # factor 被闭包捕获
return multiply
double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5))   # 10
print(triple(5))   # 15
# === 闭包实现计数器 ===
def make_counter(start: int = 0):
"""创建计数器"""
count = [start]  # 用列表包装，避免 nonlocal
def counter() -> int:
count[0] += 1
return count[0]
return counter
counter = make_counter()
print(counter())  # 1
print(counter())  # 2
print(counter())  # 3
# 使用 nonlocal
def make_counter_v2(start: int = 0):
count = start
def counter() -> int:
nonlocal count
count += 1
return count
return counter
# === 闭包做缓存 ===
def make_cache():
"""简单缓存"""
cache = {}
def cached_func(key: str, compute_fn):
if key not in cache:
cache[key] = compute_fn()
return cache[key]
return cached_func
cache = make_cache()
result = cache("heavy_calc", lambda: sum(range(1000000)))
print(result)  # 499999500000

装饰器

"""
装饰器：不修改原函数代码，添加新功能
"""
import time
import functools
# === 基本装饰器 ===
def timer(func):
"""计时装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
elapsed = time.time() - start
print(f"{func.__name__} 耗时 {elapsed:.4f}s")
return result
return wrapper
@timer
def slow_function():
"""模拟耗时操作"""
time.sleep(0.1)
return "done"
result = slow_function()  # slow_function 耗时 0.1004s
# === 带参数的装饰器 ===
def retry(max_attempts: int = 3, delay: float = 1.0):
"""重试装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
print(f"第 {attempt} 次失败: {e}")
if attempt < max_attempts:
time.sleep(delay)
raise last_exception
return wrapper
return decorator
@retry(max_attempts=3, delay=0.5)
def unreliable_api():
"""模拟不稳定的 API"""
import random
if random.random() < 0.7:
raise ConnectionError("连接失败")
return "success"
# === 日志装饰器 ===
def log_calls(func):
"""记录函数调用"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
args_repr = [repr(a) for a in args]
kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
signature = ", ".join(args_repr + kwargs_repr)
print(f"调用 {func.__name__}({signature})")
result = func(*args, **kwargs)
print(f"返回 {result!r}")
return result
return wrapper
@log_calls
def add(a: int, b: int) -> int:
return a + b
add(3, 4)
# 调用 add(3, 4)
# 返回 7
# === 缓存装饰器 ===
from functools import lru_cache
@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
"""斐波那契（自动缓存）"""
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(100))  # 秒出结果
print(fibonacci.cache_info())  # 缓存命中统计
# === 多装饰器叠加 ===
@timer
@log_calls
def process(data: str) -> str:
return data.upper()
# 等价于: timer(log_calls(process))
process("hello")

偏函数与柯里化

"""
偏函数：固定部分参数
"""
from functools import partial
# === partial ===
def power(base: int, exp: int) -> int:
return base ** exp
square = partial(power, exp=2)
cube = partial(power, exp=3)
print(square(5))  # 25
print(cube(3))    # 27
# 实用场景
import json
json_dump_cn = partial(
json.dumps,
ensure_ascii=False,
indent=2,
)
data = {"名字": "张三", "年龄": 25}
print(json_dump_cn(data))
# === 柯里化 ===
def curry(func):
"""自动柯里化"""
import inspect
params = inspect.signature(func).parameters
@functools.wraps(func)
def curried(*args, **kwargs):
if len(args) + len(kwargs) >= len(params):
return func(*args, **kwargs)
return partial(curried, *args, **kwargs)
return curried
@curry
def add_three(a: int, b: int, c: int) -> int:
return a + b + c
print(add_three(1)(2)(3))    # 6
print(add_three(1, 2)(3))    # 6
print(add_three(1, 2, 3))    # 6

本章小结

下一章：并发与异步——多线程、多进程、asyncio。