Python 函数式编程

1. 装饰器 — 基础

import functools
import time

# 基础装饰器
def timer(func):
    @functools.wraps(func)  # 保留原函数的 __name__, __doc__ 等
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f'{func.__name__} took {elapsed:.4f}s')
        return result
    return wrapper

@timer
def slow_add(a, b):
    """慢速加法"""
    time.sleep(0.1)
    return a + b

result = slow_add(1, 2)
print(f'{slow_add.__name__}: {result}')  # slow_add: 3
print(slow_add.__doc__)                   # 慢速加法

---

2. 装饰器 — 带参数

import functools

def retry(max_attempts=3, exceptions=(Exception,)):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        raise
                    print(f'Attempt {attempt} failed: {e}, retrying...')
        return wrapper
    return decorator

@retry(max_attempts=3, exceptions=(ConnectionError,))
def fetch_data(url):
    import random
    if random.random() < 0.7:
        raise ConnectionError('Network error')
    return 'data'

# 带参数的类装饰器
class Validate:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func

    def __call__(self, *args, **kwargs):
        for arg in args:
            if not isinstance(arg, (int, float)):
                raise TypeError(f'Expected number, got {type(arg)}')
        return self.func(*args, **kwargs)

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return functools.partial(self.__call__, obj)

---

3. 装饰器 — 类实现

import functools

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f'{self.func.__name__} called {self.count} times')
        return self.func(*args, **kwargs)

    def reset(self):
        self.count = 0

@CountCalls
def say_hello(name):
    print(f'Hello, {name}!')

say_hello('Alice')  # say_hello called 1 times / Hello, Alice!
say_hello('Bob')    # say_hello called 2 times / Hello, Bob!
say_hello.reset()

---

4. 装饰器 — 堆叠

import functools

def bold(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f'<b>{func(*args, **kwargs)}</b>'
    return wrapper

def italic(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f'<i>{func(*args, **kwargs)}</i>'
    return wrapper

def underline(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f'<u>{func(*args, **kwargs)}</u>'
    return wrapper

@bold
@italic
@underline
def greet(name):
    return f'Hello, {name}'

# 等价于 bold(italic(underline(greet)))
print(greet('World'))
# <b><i><u>Hello, World</u></i></b>

执行顺序（从外到内应用，从内到外执行）：

greet('World')
→ bold.wrapper: '<b>' + italic.wrapper(...) + '</b>'
  → italic.wrapper: '<i>' + underline.wrapper(...) + '</i>'
    → underline.wrapper: '<u>' + greet('World') + '</u>'
      → greet('World') = 'Hello, World'

---

5. lru_cache 实现原理

import functools
from collections import OrderedDict

# 手动实现 LRU Cache
class LRU_Cache:
    def __init__(self, maxsize=128):
        self.maxsize = maxsize
        self.cache = OrderedDict()
        self.hits = 0
        self.misses = 0

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            if key in self.cache:
                self.hits += 1
                self.cache.move_to_end(key)  # 标记为最近使用
                return self.cache[key]
            self.misses += 1
            result = func(*args, **kwargs)
            self.cache[key] = result
            if len(self.cache) > self.maxsize:
                self.cache.popitem(last=False)  # 移除最久未使用
            return result
        wrapper.cache_info = lambda: {
            'hits': self.hits,
            'misses': self.misses,
            'maxsize': self.maxsize,
            'currsize': len(self.cache),
        }
        wrapper.cache_clear = lambda: (self.cache.clear(), setattr(self, 'hits', 0), setattr(self, 'misses', 0))
        return wrapper

@LRU_Cache(maxsize=32)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(50))  # 12586269025
print(fibonacci.cache_info())

# 标准库用法
@functools.lru_cache(maxsize=256)
def expensive_calc(n):
    return sum(range(n))

@functools.cache  # Python 3.9+, 无大小限制
def simple_cache(n):
    return n * n

---

6. 闭包 — free variable 与 nonlocal

# 闭包：内部函数引用了外部函数的变量
def make_multiplier(factor):
    def multiplier(x):
        return x * factor  # factor 是自由变量 (free variable)
    return multiplier

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5))   # 10
print(triple(5))   # 15

# 查看闭包
print(double.__closure__)          # (<cell at 0x...>,)
print(double.__closure__[0].cell_contents)  # 2

nonlocal 关键字

def counter():
    count = 0
    def increment():
        nonlocal count  # 声明使用外层变量，而非创建新局部变量
        count += 1
        return count
    return increment

c = counter()
print(c())  # 1
print(c())  # 2
print(c())  # 3

# 常见陷阱 — 不使用 nonlocal
def make_accumulator():
    total = 0
    def add(n):
        # total += n  # UnboundLocalError!
        nonlocal total
        total += n
        return total
    return add

acc = make_accumulator()
print(acc(10))  # 10
print(acc(20))  # 30

---

7. 生成器 — 状态机

生成器是一种有限状态机，有 4 种状态：

# 状态：GEN_CREATED, GEN_RUNNING, GEN_SUSPENDED, GEN_CLOSED

def my_generator():
    print('Step 1')
    yield 1
    print('Step 2')
    yield 2
    print('Step 3')

gen = my_generator()
# 状态: GEN_CREATED (未开始执行)

print(next(gen))  # Step 1 / 1
# 状态: GEN_SUSPENDED (在 yield 处暂停)

print(next(gen))  # Step 2 / 2
# 状态: GEN_SUSPENDED

print(next(gen))  # Step 3 / StopIteration
# 状态: GEN_CLOSED

# 检查生成器状态
import inspect
gen = my_generator()
print(inspect.getgeneratorstate(gen))  # GEN_CREATED
next(gen)
print(inspect.getgeneratorstate(gen))  # GEN_SUSPENDED

---

8. send() 原理

# yield 既接收值，也产出值
def accumulator():
    total = 0
    value = yield total  # 第一次 yield 产出初始值 0
    while value is not None:
        total += value
        value = yield total  # 接收新值，产出累计值

gen = accumulator()
result = next(gen)       # 启动生成器, result = 0
print(result)

result = gen.send(10)    # 发送 10, result = 10
print(result)

result = gen.send(20)    # 发送 20, result = 30
print(result)

result = gen.send(30)    # 发送 30, result = 60
print(result)

throw() — 向生成器抛出异常：

def careful_generator():
    try:
        while True:
            value = yield
            print(f'Received: {value}')
    except ValueError:
        print('Got ValueError, stopping')
    finally:
        print('Generator cleanup')

gen = careful_generator()
next(gen)
gen.send('hello')
gen.send('world')
gen.throw(ValueError, 'stop!')
# Generator cleanup

---

9. yield from (PEP 380)

yield from 将生成器的迭代代理给子生成器：

# 手动实现
def chain_manual(*iterables):
    for it in iterables:
        for item in it:
            yield item

# 使用 yield from
def chain_yield_from(*iterables):
    for it in iterables:
        yield from it  # 简洁替代 for+yield

print(list(chain_yield_from([1, 2], [3, 4], [5])))
# [1, 2, 3, 4, 5]

# yield from 的关键特性：双向通道
def sub_generator():
    yield 1
    received = yield 2  # 可以接收外部 send() 的值
    yield f'Got: {received}'

def delegating_generator():
    result = yield from sub_generator()  # 代理所有 send()/throw()
    print(f'Sub returned: {result}')

gen = delegating_generator()
print(next(gen))       # 1
print(next(gen))       # 2
print(gen.send('hello'))  # Got: hello, 然后 sub_generator 返回

# 异常处理
def sub_gen():
    try:
        yield 1
        yield 2
    except ValueError:
        print('Caught in sub_gen')

def delegator():
    yield from sub_gen()

g = delegator()
next(g)
g.throw(ValueError)  # 异常自动传播到 sub_gen

---

10. 迭代器协议

class Fibonacci:
    """实现迭代器协议的斐波那契数列"""
    def __init__(self, max_count=10):
        self.max_count = max_count

    def __iter__(self):
        return self

    def __next__(self):
        if not hasattr(self, '_a'):
            self._a, self._b = 0, 1
        if self._count >= self.max_count:
            raise StopIteration
        self._count = getattr(self, '_count', 0) + 1
        result = self._a
        self._a, self._b = self._b, self._a + self._b
        return result

# 使用
for num in Fibonacci(8):
    print(num, end=' ')  # 0 1 1 2 3 5 8 13
print()

# 生成器版本（更简洁）
def fibonacci_gen(max_count=10):
    a, b = 0, 1
    for _ in range(max_count):
        yield a
        a, b = b, a + b

# 迭代协议在 Python 中的应用：
# 1. for 循环自动调用 __iter__()
# 2. 列表/字典/集合构造函数接受任何可迭代对象
# 3. unpacking (*, **) 使用迭代协议
# 4. map/filter/zip 返回迭代器

---

11. operator 模块

from operator import add, mul, itemgetter, attrgetter
from functools import reduce

# 函数式操作替代 lambda
numbers = [3, 1, 4, 1, 5, 9]

# 代替 lambda x, y: x + y
print(reduce(add, numbers))  # 23

# itemgetter — 替代 lambda x: x[key]
students = [
    {'name': 'Alice', 'grade': 90},
    {'name': 'Bob', 'grade': 85},
    {'name': 'Charlie', 'grade': 95},
]
by_grade = itemgetter('grade')
print(sorted(students, key=by_grade, reverse=True))
# [{'name': 'Charlie', ...}, {'name': 'Alice', ...}, ...]

# attrgetter — 替代 lambda x: x.attr
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

people = [Person('Alice', 30), Person('Bob', 25)]
by_name = attrgetter('name')
print(sorted(people, key=by_name))

---

12. partial vs 闭包

from functools import partial

# partial — 创建预填充参数的新函数
def power(base, exponent):
    return base ** exponent

square = partial(power, exponent=2)
cube = partial(power, exponent=3)

print(square(5))  # 25
print(cube(5))    # 125

# 闭包版本
def power_closure(exponent):
    def power(base):
        return base ** exponent
    return power

square_closure = power_closure(2)
cube_closure = power_closure(3)

# 对比
# partial:
# - 速度快（C 实现）
# - 不引入新的闭包作用域
# - 可以被 pickle（序列化）
# - 函数属性保留完整

# 闭包:
# - 更灵活（可以有任意逻辑）
# - 可以访问外部作用域的多个变量
# - 不可 pickle（除非使用全局函数）
# - 每次创建新的函数对象

# partial 不能 pickle 的闭包可以序列化
import pickle

p = partial(pow, 2)
print(pickle.loads(pickle.dumps(p)))  # OK

def make_adder(n):
    return lambda x: x + n

f = make_adder(5)
# pickle.dumps(f)  # 可能失败（取决于实现）

面试要点：partial 适合简单的参数预填充场景，闭包适合需要更多逻辑的场景。在需要序列化的场景（如多进程通信）中，优先使用 partial。