Toolz：一个函数式工具带

toolz库增强了Python的函数式编程能力，提供了一些工具，使得函数式编程更加自然。

from toolz import pipe, compose, curry, juxt

def infinite_sequence():
    num = 0
    while True:
        yield num
        num += 1

# 仅计算前 5 个值
for i, num in enumerate(infinite_sequence()):
    print(num)

if i >= 4:
        break
# 输出: 0 1 2 3 4

import itertools

# 生成无限的2的幂的序列

powers_of_two = itertools.accumulate(itertools.repeat(1), lambda x, _: x * 2)
# 取前10个元素
first_ten = list(itertools.islice(powers_of_two, 10))
print(first_ten)  # [1, 2, 4, 8, 16, 32, 64, 128, 256, 512]

# 批量处理项目
def batch(iterable, size):

it = iter(iterable)
while batch := tuple(itertools.islice(it, size)):
    yield batch

for batch in batch(range(10), 3):
    print(batch)
# 输出: (0, 1, 2), (3, 4, 5), (6, 7, 8), (9,)

from pyrsistent import pvector, pmap, s

# 不可变向量
v1 = pvector([1, 2, 3])

v2 = v1.append(4)  # 返回一个新的向量，原始向量不变
print(v1)  # pvector([1, 2, 3])
print(v2)  # pvector([1, 2, 3, 4])

# 不可变映射
m1 = pmap({'a': 1, 'b': 2})
m2 = m1.set('c', 3)
print(m1)  # pmap({'a': 1, 'b': 2})

print(m2)  # pmap({'a': 1, 'b': 2, 'c': 3})

# 不可变集合
s1 = s(1, 2, 3)
s2 = s1.add(4)
print(s1)  # pset([1, 2, 3])
print(s2)  # pset([1, 2, 3, 4])


这些结构在多线程应用中证明了其价值，我需要在不使用复杂锁机制的情况下防止竞争条件。

def process_command(command):
    match command.split():
        case ["quit"]:
            return "正在退出程序"

case ["load", filename]:
            return f"加载文件: {filename}"
        case ["save", filename]:
            return f"保存到文件: {filename}"
        case ["search", *terms]:
 return f"搜索内容: {' '.join(terms)}"
        case _:
            return "未知命令"

print(process_command("退出"))           # 退出程序
print(process_command("加载 data.txt"))  # 加载文件: data.txt
print(process_command("搜索 python 函数式编程"))
# 搜索内容: python 函数式编程

我发现模式匹配在处理结构化数据（如抽象语法树或 JSON 响应）时特别有用，使意图比嵌套的 if-else 语句更清晰。

单子与函数式错误处理

单子提供了一种结构化的方式来处理可能失败或有副作用的操作。虽然 Python 没有内置的单子，但我们可以实现像 Maybe 和 Either 这样的模式。

class Maybe:

def __init__(self, value=None):
        self.value = value

    @classmethod
    def just(cls, value):
        return cls(value)

    @classmethod
    def nothing(cls):
        return cls(None)

    def bind(self, func):
        if self.value is None:

return Maybe.nothing()
        return func(self.value)

    def __str__(self):
        if self.value is None:
            return "Nothing"
        return f"Just {self.value}"

# 使用示例
def safe_div(a, b):
    if b == 0:

return Maybe.nothing()
return Maybe.just(a / b)

def safe_sqrt(x):
if x < 0:
return Maybe.nothing()
return Maybe.just(x ** 0.5

# 链接可能失败的操作

result = Maybe.just(16).bind(lambda x: safe_div(x, 4)).bind(safe_sqrt)
print(result)  # 结果为 2.0

# 如果任何一步失败，结果为 Nothing
result = Maybe.just(16).bind(lambda x: safe_div(x, 0)).bind(safe_sqrt)

print(result)  # 无输出

Either 单子提供了关于失败的更多信息： class Either:

    class Left:
        def __init__(self, value):
 self.value = value

        def bind(self, _):
            return self

        def __str__(self):
            return f"Left({self.value})"

    class Right:
        def __init__(self, value):
            self.value = value

        def bind(self, func):
 return func(self.value)

        def __str__(self):
            return f"Right({self.value})"

# 使用示例
def div(a, b):
    if b == 0:
        return Either.Left("除以零")
    return Either.Right(a / b)
 
def sqrt(x):
    if x < 0:
        return Either.Left("无法计算负数的平方根")
    return Either.Right(x ** 0.5)

# 链式操作与错误跟踪
result = Either.Right(16).bind(lambda x: div(x, 4)).bind(sqrt)
 
print(result)  # Right(2.0)

# 带有信息性消息的错误案例
result = Either.Right(16).bind(lambda x: div(x, 0)).bind(sqrt)
print(result)  # Left(除以零)

import csv
from functools import partial, reduce
from toolz import pipe

def read_csv(filename):

with open(filename, 'r') as f:
        return list(csv.DictReader(f))

def filter_columns(data, keep_columns):
    return [{col: row[col] for col in keep_columns if col in row} for row in data]

这段代码的功能是打开一个文件并读取其内容，返回一个字典列表。`filter_columns` 函数用于过滤数据，只保留指定的列。

def filter_rows(data, predicate):
return [row for row in data if predicate(row)] def transform_column(data, column, func):
return [{**row, column: func(row[column])} for row in data]

# 处理销售数据
process_sales = partial(pipe,

partial(filter_columns, keep_columns=['date', 'product', 'price', 'quantity']),
partial(transform_column, column='price', func=float), partial(transform_column, column='quantity', func=int),

partial(filter_rows, predicate=lambda row: row['quantity'] > 0),
    partial(transform_column, column='total', func=lambda _: float(row['price']) * int(row['quantity']))
)
# 用法
processed_data = process_sales(read_csv('sales.csv'))

状态管理

def reducer(state, action):
    match action["type"]:
        case "INCREMENT":
            return {**state, "count": state["count"] + 1}
        case "DECREMENT":

return {**state, "count": state["count"] - 1}
        case "SET_COUNT":
            return {**state, "count": action["payload"]}
        case _:
            return state

# 初始状态
initial_state = {"count": 0}

# 模拟一系列操作
actions = [
    {"type": "INCREMENT"},
    {"type": "INCREMENT"},
    {"type": "SET_COUNT", "payload": 10},
    {"type": "DECREMENT"}
]

# 按顺序应用操作

final_state = reduce(reducer, actions, initial_state)
print(final_state)  # {'count': 9}

import concurrent.futures
from functools import partial

def process_chunk(process_func, chunk):
    return [process_func(item) for item in chunk]

def parallel_map(func, items, max_workers=None, chunk_size=100):
    # 创建数据块

import timeit
import functools

# 对于昂贵的计算进行备忘录优化
@functools.lru_cache(maxsize=None)
def fibonacci(n):
    if n < 2:
        return n

return fibonacci(n-1) + fibonacci(n-2)

# 比较性能
def imperative_sum():
    total = 0
    for i in range(1000):
        total += i
    return total

def functional_sum():
    return sum(range(1000))

# 测量执行时间

print(timeit.timeit(imperative_sum, number=10000))
print(timeit.timeit(functional_sum, number=10000))

这段代码的作用是使用 `timeit` 模块来测量两个函数 `imperative_sum` 和 `functional_sum` 的执行时间，每个函数执行 10000 次。 函数式编程方法有时可能会引入开销，但像记忆化这样的技术可以减轻这些成本。我发现，清晰性和可维护性的好处往往超过了微小的性能差异。 Python 的函数式工具箱在每次发布中都在不断扩展。通过将这些技术融入我的日常工作，我在代码质量和可维护性方面看到了显著的改善。函数式范式鼓励一种有纪律的编程方法，这在复杂应用中尤其有价值，因为可预测性和可测试性至关重要。 Python 中函数式编程的力量并不在于独占使用它，而在于知道何时以及如何将这些技术与 Python 的其他范式结合使用。正是这种灵活性使得 Python 成为现代软件开发中如此多才多艺的语言。