Python 纯函数编程：从理念到实战

纯函数改造：

# 最佳实践：纯函数设计
def calculate_order_total(items, discount_rate, tax_rate):
    """
    计算订单总价
    Args:
        items: 商品列表 [{'price': float, 'quantity': int}, ...]
        discount_rate: 折扣率（0-1）
        tax_rate: 税率（0-1）
    Returns:
        float: 订单总价
    """
    subtotal = sum(item['price'] * item['quantity'] for item in items)
    discount = subtotal * discount_rate
    tax = (subtotal - discount) * tax_rate
    return subtotal - discount + tax

# 优势：可预测、易测试
items = [{'price': 100, 'quantity': 2}]
total1 = calculate_order_total(items, 0.1, 0.08)
total2 = calculate_order_total(items, 0.1, 0.08)
assert total1 == total2  # 保证一致性

二、纯函数让测试变得简单

2.1 传统测试的痛点

import unittest
from datetime import datetime

# 非纯函数：依赖系统时间
def generate_report(data):
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    return f"Report generated at {timestamp}\n" + "\n".join(data)

# 测试困难
class TestReport(unittest.TestCase):
    def test_generate_report(self):
        result = generate_report(['Line 1', 'Line 2'])
        # 如何验证？时间戳每次都不同
        self.assertIn('Report generated at', result)
        # 只能做模糊匹配，无法精确验证

2.2 纯函数的测试优势

from datetime import datetime

# 纯函数改造：依赖注入
def generate_report_pure(data, timestamp):
    """生成报告（纯函数版本）"""
    return f"Report generated at {timestamp}\n" + "\n".join(data)

# 测试简单明了
class TestReportPure(unittest.TestCase):
    def test_generate_report(self):
        data = ['Line 1', 'Line 2']
        timestamp = '2024-01-01 10:00:00'
        result = generate_report_pure(data, timestamp)
        expected = "Report generated at 2024-01-01 10:00:00\nLine 1\nLine 2"
        self.assertEqual(result, expected)

    def test_empty_data(self):
        result = generate_report_pure([], '2024-01-01 10:00:00')
        self.assertEqual(result, "Report generated at 2024-01-01 10:00:00\n")

# 运行测试
if __name__ == '__main__':
    unittest.main()

2.3 实战案例：数据处理管道

from typing import List, Callable

# 纯函数组件
def filter_valid_emails(emails: List[str]) -> List[str]:
    """过滤有效邮箱"""
    return [email for email in emails if '@' in email and '.' in email.split('@')[1]]

def normalize_emails(emails: List[str]) -> List[str]:
    """标准化邮箱格式"""
    return [email.lower().strip() for email in emails]

def deduplicate(items: List[str]) -> List[str]:
    """去重"""
    return list(dict.fromkeys(items))

# 函数组合（纯函数的强大之处）
def compose(*functions: Callable) -> Callable:
    """组合多个函数"""
    def inner(data):
        result = data
        for func in functions:
            result = func(result)
        return result
    return inner

# 构建数据处理管道
email_pipeline = compose(
    normalize_emails,
    filter_valid_emails,
    deduplicate
)

# 测试
def test_email_pipeline():
    raw_data = ['[email protected]', '[email protected]', 'invalid-email', ' [email protected] ', '[email protected]']
    result = email_pipeline(raw_data)
    expected = ['[email protected]', '[email protected]']
    assert result == expected
    print("测试通过！")

test_email_pipeline()

三、纯函数与并发：天作之合

3.1 并发编程的挑战

import threading

# 非纯函数：线程不安全
balance = 1000
def withdraw(amount):
    global balance
    if balance >= amount:
        # 模拟处理延迟
        import time
        time.sleep(0.001)
        balance -= amount
    return True
return False

# 并发问题演示
threads = [threading.Thread(target=withdraw, args=(100,)) for _ in range(15)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"剩余余额：{balance}")
# 结果不可预测！可能出现负数

3.2 纯函数实现线程安全

from dataclasses import dataclass
from typing import Tuple
from concurrent.futures import ThreadPoolExecutor

@dataclass(frozen=True)
# 不可变数据结构
class Account:
    balance: float

def withdraw(self, amount: float) -> Tuple['Account', bool]:
    """纯函数：返回新状态，不修改原对象"""
    if self.balance >= amount:
        return Account(self.balance - amount), True
    return self, False

# 并发安全的实现
def process_withdrawal(account: Account, amount: float) -> Account:
    new_account, success = account.withdraw(amount)
    return new_account if success else account

# 使用不可变数据结构 + 纯函数
initial_account = Account(balance=1000)
# 串行处理（或使用消息队列）
withdrawals = [100] * 15
final_account = initial_account
for amount in withdrawals:
    final_account = process_withdrawal(final_account, amount)
print(f"最终余额：{final_account.balance}")
# 结果可预测：-500

3.3 实战：并行数据处理

from concurrent.futures import ProcessPoolExecutor
from typing import List
import time

# 纯函数：CPU 密集型任务
def process_chunk(numbers: List[int]) -> int:
    """计算列表中质数的个数"""
    def is_prime(n):
        if n < 2:
            return False
        for i in range(2, int(n ** 0.5) + 1):
            if n % i == 0:
                return False
        return True
    return sum(1 for num in numbers if is_prime(num))

# 性能对比
def sequential_processing(data: List[int]) -> int:
    """串行处理"""
    return process_chunk(data)

def parallel_processing(data: List[int], num_workers: int = 4) -> int:
    """并行处理（纯函数天然支持）"""
    chunk_size = len(data) // num_workers
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        results = executor.map(process_chunk, chunks)
    return sum(results)

# 测试
if __name__ == '__main__':
    test_data = list(range(1, 100000))
    start = time.time()
    result1 = sequential_processing(test_data)
    time1 = time.time() - start
    start = time.time()
    result2 = parallel_processing(test_data)
    time2 = time.time() - start
    print(f"串行处理：{result1} 个质数，耗时 {time1:.2f}秒")
    print(f"并行处理：{result2} 个质数，耗时 {time2:.2f}秒")
    print(f"性能提升：{time1/time2:.2f}x")

四、实践技巧与常见陷阱

4.1 不可变数据结构的运用

from typing import NamedTuple, List

# 使用 NamedTuple 创建不可变对象
class Point(NamedTuple):
    x: float
    y: float

def move(self, dx: float, dy: float) -> 'Point':
    """返回新位置"""
    return Point(self.x + dx, self.y + dy)

# 使用 frozenset 代替 set
def unique_intersection(list1: List[int], list2: List[int]) -> frozenset:
    """纯函数：计算两个列表的交集"""
    return frozenset(list1) & frozenset(list2)

4.2 避免隐藏的副作用

# 陷阱：看似纯函数，实则有副作用
def append_item(items: List[int], item: int) -> List[int]:
    items.append(item)  # 修改了传入参数！
    return items

original = [1, 2, 3]
result = append_item(original, 4)
print(original)  # [1, 2, 3, 4] 被修改了！

# 正确做法：创建新列表
def append_item_pure(items: List[int], item: int) -> List[int]:
    return items + [item]

# 或 [*items, item]
original = [1, 2, 3]
result = append_item_pure(original, 4)
print(original)  # [1, 2, 3] 保持不变
print(result)    # [1, 2, 3, 4]

4.3 性能与纯函数的平衡

# 场景：大数据处理
def process_large_dataset(data: List[dict]) -> List[dict]:
    """
    纯函数方式：适合中小规模数据
    """
    return [{**item, 'processed': True, 'score': item['value'] * 2} for item in data]

# 优化：使用生成器（保持纯函数特性）
def process_large_dataset_lazy(data: List[dict]):
    """
    惰性求值：内存友好
    """
    for item in data:
        yield {**item, 'processed': True, 'score': item['value'] * 2}

# 使用示例
large_data = [{'value': i} for i in range(1000000)]

# 方法一：内存占用高
# result = process_large_dataset(large_data)

# 方法二：按需计算
for processed_item in process_large_dataset_lazy(large_data):
    # 逐个处理，内存占用低
    pass

五、总结与展望

纯函数不是银弹，但它为我们提供了一种强大的编程思维：通过约束来获得自由。当你拥抱纯函数理念，你会发现：

• 测试变得轻松：无需 mock 复杂环境，输入输出一目了然
• 并发更安全：天然线程安全，轻松实现并行计算
• 代码更易维护：函数职责清晰，重构时信心十足
• bug 更少：可预测性强，边界情况易于控制

实践建议

1. 从小处着手：先将工具函数改造为纯函数
2. 识别边界：I/O 操作放在系统边缘，核心逻辑保持纯净
3. 善用工具：dataclasses、NamedTuple、frozenset 是你的好帮手
4. 性能权衡：必要时使用生成器和惰性求值