itertools 与 functools 完全指南
itertools 和 functools 是 Python 标准库中两个专注于函数式编程的模块。itertools 提供高效的迭代器工具,functools 提供高阶函数和函数操作工具。两者结合使用能写出简洁、高效、声明式的 Python 代码。 无限迭代器不会自动停止,必须配合 islice、takewhile 或 break 使用,否则会进入死循环。 从起始值开始无限递增计数。 无限循环遍历可迭代对象的元素。 注意:cycle 会将整个可迭代对象的内容缓存在内存中,对大型序列要谨慎使用。 重复输出同一个值,可指定次数或无限重复。 将多个可迭代对象首
官方文档:https://docs.python.org/3/library/itertools.html | https://docs.python.org/3/library/functools.html
适用版本:Python 3.12(2026-05-08 核实)
itertools 和 functools 是 Python 标准库中两个专注于函数式编程的模块。itertools 提供高效的迭代器工具,functools 提供高阶函数和函数操作工具。两者结合使用能写出简洁、高效、声明式的 Python 代码。
itertools
无限迭代器
无限迭代器不会自动停止,必须配合 islice、takewhile 或 break 使用,否则会进入死循环。
count
从起始值开始无限递增计数。
import itertools
itertools.count(start=0, step=1)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| start | int / float | 0 | 起始值 |
| step | int / float | 1 | 步长,可为负数或小数 |
import itertools
# 基础用法:生成有限序列
for n in itertools.islice(itertools.count(10, 2), 5):
print(n) # 10, 12, 14, 16, 18
# 浮点步长
counter = itertools.count(0.0, 0.5)
print(list(itertools.islice(counter, 4))) # [0.0, 0.5, 1.0, 1.5]
# 与 zip 结合为序列添加编号(比 enumerate 更灵活,可自定义步长)
data = ['a', 'b', 'c']
numbered = list(zip(itertools.count(1), data))
print(numbered) # [(1, 'a'), (2, 'b'), (3, 'c')]
cycle
无限循环遍历可迭代对象的元素。
itertools.cycle(iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 要循环的可迭代对象,内部会缓存所有元素 |
import itertools
# 循环颜色标签
colors = itertools.cycle(['red', 'green', 'blue'])
result = [next(colors) for _ in range(7)]
print(result) # ['red', 'green', 'blue', 'red', 'green', 'blue', 'red']
# 轮询负载均衡
servers = itertools.cycle(['server1', 'server2', 'server3'])
requests = ['req1', 'req2', 'req3', 'req4', 'req5']
assignments = [(req, next(servers)) for req in requests]
注意:cycle 会将整个可迭代对象的内容缓存在内存中,对大型序列要谨慎使用。
repeat
重复输出同一个值,可指定次数或无限重复。
itertools.repeat(object, times=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| object | any | 必填 | 要重复的对象 |
| times | int | None | 重复次数,None 表示无限 |
import itertools
# 有限重复
print(list(itertools.repeat(10, 3))) # [10, 10, 10]
# 与 map/starmap 配合传递固定参数
result = list(map(pow, range(5), itertools.repeat(2)))
print(result) # [0, 1, 4, 9, 16]
有限迭代器
chain
将多个可迭代对象首尾相连,形成一个连续序列。
itertools.chain(*iterables)
itertools.chain.from_iterable(iterable) # 接受嵌套可迭代对象
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| *iterables | iterable | 必填 | 一个或多个可迭代对象 |
import itertools
# 基础连接
result = list(itertools.chain([1, 2], [3, 4], [5]))
print(result) # [1, 2, 3, 4, 5]
# from_iterable:扁平化嵌套列表
nested = [[1, 2], [3, 4], [5, 6]]
flat = list(itertools.chain.from_iterable(nested))
print(flat) # [1, 2, 3, 4, 5, 6]
# 合并多个字典的键(Python 3.9+ 有更好的方式,但 chain 仍适用于迭代场景)
dicts = [{'a': 1}, {'b': 2}, {'c': 3}]
all_keys = list(itertools.chain.from_iterable(d.keys() for d in dicts))
islice
对迭代器进行切片,类似列表切片但不支持负数索引。
itertools.islice(iterable, stop)
itertools.islice(iterable, start, stop, step=1)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 要切片的可迭代对象 |
| start | int | 0 | 起始位置(含),只传一个整数参数时为 stop |
| stop | int / None | 必填 | 结束位置(不含),None 表示到末尾 |
| step | int | 1 | 步长,不支持负数 |
import itertools
# 取前 5 个
result = list(itertools.islice(range(100), 5))
print(result) # [0, 1, 2, 3, 4]
# 跳过前 3 个,取接下来的 4 个
result = list(itertools.islice(range(10), 3, 7))
print(result) # [3, 4, 5, 6]
# 每隔一个取一个
result = list(itertools.islice(range(10), 0, None, 2))
print(result) # [0, 2, 4, 6, 8]
# 读取大文件的前 100 行
def read_first_lines(filename, n=100):
with open(filename) as f:
return list(itertools.islice(f, n))
takewhile
在条件为真时持续取值,一旦条件为假立即停止(不再检查后续元素)。
itertools.takewhile(predicate, iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| predicate | callable | 必填 | 判断函数,接受一个元素,返回布尔值 |
| iterable | iterable | 必填 | 要过滤的可迭代对象 |
import itertools
result = list(itertools.takewhile(lambda x: x < 5, [1, 2, 3, 4, 5, 1, 2]))
print(result) # [1, 2, 3, 4]
# 注意:遇到 5 后停止,后面的 1, 2 也不会被取到
# 读取日志直到遇到 ERROR
def read_until_error(log_lines):
return list(itertools.takewhile(
lambda line: 'ERROR' not in line,
log_lines
))
dropwhile
跳过满足条件的前缀元素,条件首次为假后返回剩余所有元素。
itertools.dropwhile(predicate, iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| predicate | callable | 必填 | 判断函数,接受一个元素,返回布尔值 |
| iterable | iterable | 必填 | 要过滤的可迭代对象 |
import itertools
result = list(itertools.dropwhile(lambda x: x < 5, [1, 2, 3, 6, 4, 1]))
print(result) # [6, 4, 1]
# 条件首次为假(遇到 6)后,后续所有元素(包括 4, 1)都保留
# 跳过文件头部注释行
def skip_comments(lines):
return list(itertools.dropwhile(
lambda line: line.startswith('#'),
lines
))
filterfalse
返回使 predicate 为假的元素(与内置 filter 相反)。
itertools.filterfalse(predicate, iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| predicate | callable / None | 必填 | 判断函数,None 时过滤所有假值 |
| iterable | iterable | 必填 | 要过滤的可迭代对象 |
import itertools
# 过滤掉偶数(保留奇数)
result = list(itertools.filterfalse(lambda x: x % 2 == 0, range(10)))
print(result) # [1, 3, 5, 7, 9]
# predicate 为 None 时过滤假值(等价于 filter(None, iterable) 的反面)
result = list(itertools.filterfalse(None, [0, 1, '', 'a', None, True]))
print(result) # [0, '', None]
compress
根据选择器序列过滤数据,选择器为真则保留对应元素。
itertools.compress(data, selectors)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| data | iterable | 必填 | 数据序列 |
| selectors | iterable | 必填 | 布尔选择器序列,长度可与 data 不同,以较短者为准 |
import itertools
data = ['a', 'b', 'c', 'd', 'e']
selectors = [1, 0, 1, 0, 1]
result = list(itertools.compress(data, selectors))
print(result) # ['a', 'c', 'e']
# 根据条件列表筛选
scores = [85, 42, 91, 67, 73]
passed = [s >= 60 for s in scores]
names = ['Alice', 'Bob', 'Carol', 'Dave', 'Eve']
result = list(itertools.compress(names, passed))
print(result) # ['Alice', 'Carol', 'Dave', 'Eve']
starmap
将可迭代对象中的每个元素(元组)展开后作为参数传给函数。
itertools.starmap(function, iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| function | callable | 必填 | 接受多个参数的函数 |
| iterable | iterable | 必填 | 元素为元组或可解包序列的可迭代对象 |
import itertools
import operator
# 基础用法
pairs = [(2, 3), (4, 2), (5, 3)]
result = list(itertools.starmap(pow, pairs))
print(result) # [8, 16, 125]
# 与 operator 结合
data = [(1, 2), (3, 4), (5, 6)]
result = list(itertools.starmap(operator.add, data))
print(result) # [3, 7, 11]
# starmap vs map 的区别
# map(func, [1, 2, 3]) 将每个元素作为单个参数传入
# starmap(func, [(1,2), (3,4)]) 将每个元组展开后传入
zip_longest
类似内置 zip,但以最长序列为准,短序列用 fillvalue 填充。
itertools.zip_longest(*iterables, fillvalue=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| *iterables | iterable | 必填 | 一个或多个可迭代对象 |
| fillvalue | any | None | 短序列耗尽后的填充值 |
import itertools
a = [1, 2, 3, 4, 5]
b = ['a', 'b', 'c']
result = list(itertools.zip_longest(a, b, fillvalue='-'))
print(result) # [(1, 'a'), (2, 'b'), (3, 'c'), (4, '-'), (5, '-')]
# 合并两列数据,长度不一时用默认值补齐
headers = ['name', 'age', 'city']
values = ['Alice', 25]
row = dict(itertools.zip_longest(headers, values, fillvalue='N/A'))
print(row) # {'name': 'Alice', 'age': 25, 'city': 'N/A'}
pairwise(Python 3.10+)
返回相邻元素对,等价于 zip(iterable, iterable[1:])。
itertools.pairwise(iterable)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 要生成相邻对的可迭代对象,长度小于 2 时返回空迭代器 |
import itertools
result = list(itertools.pairwise([1, 2, 3, 4, 5]))
print(result) # [(1, 2), (2, 3), (3, 4), (4, 5)]
# 计算相邻差值
prices = [100, 105, 98, 110, 107]
changes = [b - a for a, b in itertools.pairwise(prices)]
print(changes) # [5, -7, 12, -3]
# 检测排序
def is_sorted(seq):
return all(a <= b for a, b in itertools.pairwise(seq))
print(is_sorted([1, 2, 3, 4])) # True
print(is_sorted([1, 3, 2, 4])) # False
组合迭代器
product
计算多个可迭代对象的笛卡尔积,等价于嵌套 for 循环。
itertools.product(*iterables, repeat=1)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| *iterables | iterable | 必填 | 参与笛卡尔积的可迭代对象 |
| repeat | int | 1 | 将单个可迭代对象与自身做笛卡尔积的次数 |
import itertools
# 基础笛卡尔积
result = list(itertools.product([1, 2], ['a', 'b']))
print(result) # [(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')]
# repeat 参数:等价于 product(iterable, iterable)
result = list(itertools.product([0, 1], repeat=3))
print(result)
# [(0,0,0),(0,0,1),(0,1,0),(0,1,1),(1,0,0),(1,0,1),(1,1,0),(1,1,1)]
# 这是所有 3 位二进制数
# 生成测试参数组合
sizes = ['small', 'large']
colors = ['red', 'blue']
materials = ['cotton', 'silk']
skus = list(itertools.product(sizes, colors, materials))
permutations
生成所有排列(有序,不重复选取)。
itertools.permutations(iterable, r=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 元素来源 |
| r | int | None | 排列长度,None 时等于元素总数 |
import itertools
# 全排列
result = list(itertools.permutations([1, 2, 3]))
print(result)
# [(1,2,3),(1,3,2),(2,1,3),(2,3,1),(3,1,2),(3,2,1)]
# 指定长度的排列
result = list(itertools.permutations('ABCD', 2))
print(len(result)) # 12 = 4 * 3
# 生成所有可能的密码(实际应用要控制长度)
digits = '0123456789'
pin_count = sum(1 for _ in itertools.permutations(digits, 4))
print(pin_count) # 5040
combinations
生成所有组合(无序,不重复选取)。
itertools.combinations(iterable, r)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 元素来源 |
| r | int | 必填 | 组合长度 |
import itertools
result = list(itertools.combinations([1, 2, 3, 4], 2))
print(result)
# [(1,2),(1,3),(1,4),(2,3),(2,4),(3,4)]
# 从候选人中选出委员会
candidates = ['Alice', 'Bob', 'Carol', 'Dave']
committees = list(itertools.combinations(candidates, 3))
print(len(committees)) # 4 = C(4,3)
# 计算所有股票对的相关性
tickers = ['AAPL', 'GOOGL', 'MSFT', 'AMZN']
pairs = list(itertools.combinations(tickers, 2))
# [('AAPL', 'GOOGL'), ('AAPL', 'MSFT'), ...]
combinations_with_replacement
生成允许重复元素的组合(无序,可重复选取)。
itertools.combinations_with_replacement(iterable, r)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 元素来源 |
| r | int | 必填 | 组合长度 |
import itertools
result = list(itertools.combinations_with_replacement([1, 2, 3], 2))
print(result)
# [(1,1),(1,2),(1,3),(2,2),(2,3),(3,3)]
# 与 combinations 的区别:
# combinations('ABC', 2) -> AB, AC, BC(不含 AA, BB, CC)
# combinations_with_replacement('ABC', 2) -> AA, AB, AC, BB, BC, CC
# 多项式展开中的指数组合
vars_ = ['x', 'y', 'z']
degree2_terms = list(itertools.combinations_with_replacement(vars_, 2))
# [('x','x'),('x','y'),('x','z'),('y','y'),('y','z'),('z','z')]
groupby:按键分组
groupby 对连续相同键值的元素进行分组,返回 (key, group) 对。
itertools.groupby(iterable, key=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| iterable | iterable | 必填 | 要分组的可迭代对象 |
| key | callable | None | 提取分组键的函数,None 时元素本身作为键 |
import itertools
from operator import itemgetter
# 正确用法:先排序,再分组
data = [
{'name': 'Alice', 'dept': 'Engineering'},
{'name': 'Bob', 'dept': 'Marketing'},
{'name': 'Carol', 'dept': 'Engineering'},
{'name': 'Dave', 'dept': 'Marketing'},
{'name': 'Eve', 'dept': 'Engineering'},
]
# 必须先按分组键排序
data.sort(key=itemgetter('dept'))
for dept, members in itertools.groupby(data, key=itemgetter('dept')):
names = [m['name'] for m in members]
print(f"{dept}: {names}")
# Engineering: ['Alice', 'Carol', 'Eve']
# Marketing: ['Bob', 'Dave']
# 统计连续字符的出现次数(RLE 编码)
def run_length_encode(s):
return [(k, len(list(g))) for k, g in itertools.groupby(s)]
print(run_length_encode('aaabbbccddddee'))
# [('a', 3), ('b', 3), ('c', 2), ('d', 4), ('e', 2)]
分组原理:groupby 内部维护一个当前键值,每次调用 next() 时与上一个键值比较,若不同则开始新分组。因此数据必须事先按同一键排好序,否则相同键值的元素会被分到不同组。
重要陷阱:从 groupby 返回的 group 迭代器是共享底层迭代器的,一旦进入下一个分组,上一个 group 迭代器就会失效。如需保留分组数据,必须立即转换为列表:
# 错误:先收集 group 再使用
groups = [(k, g) for k, g in itertools.groupby(data, key=itemgetter('dept'))]
for k, g in groups:
print(list(g)) # 全部为空!
# 正确:立即消费 group
groups = [(k, list(g)) for k, g in itertools.groupby(data, key=itemgetter('dept'))]
functools
partial:固定参数
partial 固定函数的部分参数,返回一个新的可调用对象。
functools.partial(func, /, *args, **kwargs)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| func | callable | 必填 | 原始函数 |
| *args | any | - | 预先固定的位置参数 |
| **kwargs | any | - | 预先固定的关键字参数 |
from functools import partial
# 固定位置参数
def power(base, exp):
return base ** exp
square = partial(power, exp=2)
cube = partial(power, exp=3)
print(square(5)) # 25
print(cube(3)) # 27
# 固定关键字参数
import json
pretty_print = partial(json.dumps, indent=4, ensure_ascii=False)
data = {'name': '张三', 'age': 30}
print(pretty_print(data))
# 用于 sorted/map 等高阶函数
from functools import partial
def multiply(x, factor):
return x * factor
double = partial(multiply, factor=2)
result = list(map(double, [1, 2, 3, 4]))
print(result) # [2, 4, 6, 8]
partial 与 lambda 的区别:
| 特性 | partial | lambda |
|---|---|---|
| 可序列化(pickle) | 是 | 否(通常) |
| 可读性 | 有明确名称 | 匿名,语义不明 |
| 参数检查 | 调用时检查 | 调用时检查 |
| 内省(.func, .args) | 支持 | 不支持 |
| 适用场景 | 固定已有函数的参数 | 简单一次性变换 |
# 查看 partial 对象的属性
p = partial(power, exp=2)
print(p.func) # <function power at ...>
print(p.args) # ()
print(p.keywords) # {'exp': 2}
reduce:折叠操作
对序列进行累积运算,将二元函数从左到右依次应用到所有元素。
functools.reduce(function, iterable, initializer=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| function | callable | 必填 | 接受两个参数的二元函数 |
| iterable | iterable | 必填 | 要处理的序列 |
| initializer | any | None | 初始值,序列为空时作为结果,也是第一次调用的第一个参数 |
from functools import reduce
import operator
# 计算阶乘
result = reduce(operator.mul, range(1, 6))
print(result) # 120
# 展开嵌套列表(浅层)
nested = [[1, 2], [3, 4], [5, 6]]
flat = reduce(operator.add, nested)
print(flat) # [1, 2, 3, 4, 5, 6]
# 带初始值:防止空序列报错
result = reduce(operator.add, [], 0)
print(result) # 0(若无 initializer,空序列会抛 TypeError)
# 构建嵌套字典路径
def get_nested(data, keys):
return reduce(lambda d, k: d[k], keys, data)
config = {'database': {'host': 'localhost', 'port': 5432}}
print(get_nested(config, ['database', 'host'])) # 'localhost'
# 执行函数管道
def pipeline(*functions):
return reduce(lambda f, g: lambda x: g(f(x)), functions)
process = pipeline(str.strip, str.lower, lambda s: s.replace(' ', '_'))
print(process(' Hello World ')) # 'hello_world'
lru_cache / cache:缓存装饰器
lru_cache 使用最近最少使用策略缓存函数调用结果,cache(Python 3.9+)是 lru_cache(maxsize=None) 的简写。
@functools.lru_cache(maxsize=128, typed=False)
@functools.cache # Python 3.9+,等价于 lru_cache(maxsize=None)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| maxsize | int / None | 128 | 缓存条目上限,None 表示无限缓存(cache 默认行为);设为 2 的幂次性能最佳 |
| typed | bool | False | True 时对不同类型的参数分别缓存,如 f(3) 和 f(3.0) 视为不同调用 |
from functools import lru_cache, cache
# 经典示例:斐波那契数列
@lru_cache(maxsize=None)
def fib(n):
if n < 2:
return n
return fib(n - 1) + fib(n - 2)
print(fib(100)) # 354224848179261915075,瞬间完成
# 查看缓存状态
print(fib.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=None, currsize=101)
# 清除缓存
fib.cache_clear()
print(fib.cache_info())
# CacheInfo(hits=0, misses=0, maxsize=None, currsize=0)
# typed=True 示例
@lru_cache(maxsize=128, typed=True)
def process(x):
print(f"Computing for {x!r}")
return x * 2
process(3) # 触发计算
process(3.0) # typed=True 时视为不同参数,再次触发计算
process(3) # 命中缓存
# Python 3.9+ 使用 cache(无大小限制,适合纯计算)
@cache
def factorial(n):
return n * factorial(n - 1) if n else 1
cache_info() 返回值说明:
| 字段 | 说明 |
|---|---|
| hits | 缓存命中次数 |
| misses | 缓存未命中次数(实际执行函数的次数) |
| maxsize | 设置的最大缓存条目数 |
| currsize | 当前缓存条目数 |
使用限制:所有参数必须是可哈希的(hashable)。不可哈希类型(列表、字典、集合)作为参数时会抛出 TypeError。
@lru_cache
def process_list(data): # 错误!列表不可哈希
pass
process_list([1, 2, 3]) # TypeError: unhashable type: 'list'
# 解决方案:转换为元组
@lru_cache
def process_list(data: tuple):
pass
process_list(tuple([1, 2, 3])) # 正常工作
wraps:保留函数元信息
在定义装饰器时使用 wraps,确保被装饰函数的 __name__、__doc__、__annotations__ 等属性不被覆盖。
functools.wraps(wrapped, assigned=WRAPPER_ASSIGNMENTS, updated=WRAPPER_UPDATES)
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| wrapped | callable | 必填 | 被装饰的原始函数 |
| assigned | tuple | WRAPPER_ASSIGNMENTS | 要复制的属性名元组,默认含 __module__、__name__、__qualname__、__annotations__、__doc__ |
| updated | tuple | WRAPPER_UPDATES | 要更新(合并)的属性名元组,默认含 __dict__ |
from functools import wraps
def timer(func):
@wraps(func) # 没有这行,func.__name__ 会变成 'wrapper'
def wrapper(*args, **kwargs):
import time
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 calculate(n):
"""计算 1 到 n 的和"""
return sum(range(n))
print(calculate.__name__) # 'calculate'(而非 'wrapper')
print(calculate.__doc__) # '计算 1 到 n 的和'
# wraps 还会复制 __wrapped__ 属性,方便调试时访问原始函数
print(calculate.__wrapped__) # <function calculate at ...>
更多装饰器相关内容参见 装饰器与函数高级。
total_ordering:自动补全比较方法
只需定义 __eq__ 和一个比较方法(__lt__、__le__、__gt__、__ge__ 之一),total_ordering 自动推导其余三个。
@functools.total_ordering
无参数,直接作为类装饰器使用。
from functools import total_ordering
@total_ordering
class Student:
def __init__(self, name, gpa):
self.name = name
self.gpa = gpa
def __eq__(self, other):
return self.gpa == other.gpa
def __lt__(self, other):
return self.gpa < other.gpa
# total_ordering 自动生成 __le__, __gt__, __ge__
alice = Student('Alice', 3.8)
bob = Student('Bob', 3.5)
print(alice > bob) # True(自动生成)
print(alice >= bob) # True(自动生成)
print(alice <= bob) # False(自动生成)
# 可用于 sorted、min、max
students = [Student('C', 3.2), Student('A', 3.9), Student('B', 3.5)]
top = max(students)
print(top.name) # 'A'
注意:total_ordering 有轻微性能开销(通过反射推导),对性能敏感场景建议手动实现所有比较方法。
singledispatch:函数重载
根据第一个参数的类型分发到不同实现,实现类似其他语言的函数重载。
@functools.singledispatch
from functools import singledispatch
@singledispatch
def process(value):
"""默认实现(fallback)"""
raise TypeError(f"Unsupported type: {type(value)}")
@process.register(int)
def process_int(value):
return f"Integer: {value * 2}"
@process.register(str)
def process_str(value):
return f"String: {value.upper()}"
@process.register(list)
def process_list(value):
return f"List with {len(value)} items"
print(process(42)) # 'Integer: 84'
print(process('hello')) # 'String: HELLO'
print(process([1, 2, 3])) # 'List with 3 items'
# Python 3.7+ 支持类型注解语法
@process.register
def process_float(value: float):
return f"Float: {value:.2f}"
print(process(3.14)) # 'Float: 3.14'
# 查看注册的实现
print(process.dispatch(int)) # <function process_int ...>
print(process.registry) # 所有已注册类型的字典
类方法版本使用 singledispatchmethod(Python 3.8+):
from functools import singledispatchmethod
class Converter:
@singledispatchmethod
def convert(self, value):
raise TypeError(f"Cannot convert {type(value)}")
@convert.register(int)
def _(self, value):
return str(value)
@convert.register(str)
def _(self, value):
return int(value)
cached_property:惰性属性
将方法转换为惰性计算的属性:首次访问时计算并缓存到实例 __dict__,后续访问直接读取缓存。
@functools.cached_property
无参数,直接作为方法装饰器使用。
from functools import cached_property
class Circle:
def __init__(self, radius):
self.radius = radius
@cached_property
def area(self):
import math
print("Computing area...")
return math.pi * self.radius ** 2
@cached_property
def circumference(self):
import math
return 2 * math.pi * self.radius
c = Circle(5)
print(c.area) # 输出 "Computing area..." 然后输出结果
print(c.area) # 直接返回缓存值,不再打印 "Computing area..."
# 缓存存储在实例的 __dict__ 中
print('area' in c.__dict__) # True
# 删除缓存(下次访问时重新计算)
del c.__dict__['area']
print(c.area) # 再次输出 "Computing area..."
与 property 的区别:
| 特性 | property | cached_property |
|---|---|---|
| 每次访问 | 重新执行 getter | 首次后读缓存 |
| 适合场景 | 需要实时计算、有 setter | 计算代价大、结果不变 |
| 线程安全 | 是(不存储状态) | 否(需额外加锁) |
| 实例 dict | 不写入 | 写入同名键 |
注意:cached_property 要求类不能定义 __slots__(或 __slots__ 中包含该属性名),且类的 __dict__ 必须是可写的。
踩坑与注意事项
1. groupby 不预排序
import itertools
data = [1, 2, 1, 2, 1] # 未排序
for key, group in itertools.groupby(data):
print(key, list(group))
# 输出:
# 1 [1]
# 2 [2]
# 1 [1]
# 2 [2]
# 1 [1]
# 期望的 {1: [1,1,1], 2: [2,2]} 分组需要先排序!
data.sort()
for key, group in itertools.groupby(data):
print(key, list(group))
# 1 [1, 1, 1]
# 2 [2, 2]
2. lru_cache 不能用于不可哈希参数
from functools import lru_cache
@lru_cache(maxsize=128)
def compute(data):
return sum(data)
compute([1, 2, 3]) # TypeError: unhashable type: 'list'
compute({'a': 1}) # TypeError: unhashable type: 'dict'
# 解决方案 1:转换类型
compute(tuple([1, 2, 3])) # 可行
# 解决方案 2:使用自定义缓存装饰器
# 解决方案 3:对于字典,可用 frozenset(d.items()) 作为缓存键
3. partial 与 lambda 的序列化差异
import pickle
from functools import partial
def add(x, y):
return x + y
# partial 可以 pickle
add5 = partial(add, 5)
serialized = pickle.dumps(add5)
restored = pickle.loads(serialized)
print(restored(3)) # 8
# lambda 通常不能 pickle
add5_lambda = lambda x: add(5, x)
pickle.dumps(add5_lambda) # AttributeError 或 PicklingError
4. cycle 的内存占用
import itertools
import sys
# cycle 会缓存所有元素
big_list = list(range(1_000_000))
cycled = itertools.cycle(big_list)
# 此时 cycled 内部缓存了 1_000_000 个元素
# 对于大型数据源,考虑使用生成器手动实现 cycle
def cycle_gen(iterable):
saved = []
for element in iterable:
yield element
saved.append(element)
while saved:
yield from saved
5. lru_cache 装饰实例方法的内存泄漏
from functools import lru_cache
class MyClass:
@lru_cache(maxsize=128) # 危险!
def expensive_method(self, x):
return x * 2
# lru_cache 以 self 作为缓存键的一部分
# 只要缓存存在,就会持有对实例的强引用,导致实例无法被垃圾回收
# 解决方案:使用 cached_property(无参数方法)或在类外部管理缓存
# 或使用 methodtools 库的 lru_cache
最佳实践
itertools.chain 替代列表合并:list1 + list2 会创建新列表,itertools.chain(list1, list2) 惰性迭代,节省内存;处理大数据流时首选 chain:
from itertools import chain
for item in chain(list1, list2, list3):
process(item) # 不创建中间列表
functools.partial 固定参数创建特化函数:比 lambda 更清晰,且支持 functools.update_wrapper 保留原函数元信息:
from functools import partial
int_from_hex = partial(int, base=16)
int_from_hex('ff') # 255
itertools.groupby 前必须排序:groupby 只对连续相同 key 的元素分组,未排序则每次 key 变化都产生新组:
# 正确:先按 key 排序
for key, group in groupby(sorted(items, key=keyfn), key=keyfn):
...
functools.cache(Python 3.9+)替代 lru_cache(None):cache 是 lru_cache(maxsize=None) 的简写,语义更明确,适合结果数量有限的纯函数。
itertools.islice 安全截取无限迭代器:从生成器中取前 N 个元素,不触发完整求值:
from itertools import islice, count
first_10_evens = list(islice((x for x in count() if x % 2 == 0), 10))
常见陷阱
陷阱:lru_cache 缓存可变参数类型会报错
现象: 给 @lru_cache 装饰的函数传 list 或 dict 参数时,报 TypeError: unhashable type: 'list'。
原因: lru_cache 用参数作为缓存键,键必须可哈希,可变类型不可哈希。
解决: 将列表转为 tuple 传入,或将 dict 序列化为 frozenset(items()),或改用 functools.cache 配合 __hash__ 支持的不可变包装类型。
陷阱:itertools.cycle 生成无限序列导致内存增长
现象: cycle(iterable) 处理大型 iterable 时内存不断增长。
原因: cycle 必须缓存所有元素才能循环,传入大型迭代器会将其完整加载到内存。
解决: 若 iterable 是已知的小集合(如 ['a', 'b', 'c']),直接传字面量;若是大数据流,用自定义生成器手动实现循环逻辑。
陷阱:functools.reduce 对空序列报错
现象: reduce(op, []) 报 TypeError: reduce() of empty iterable with no initial value。
原因: reduce 无法对空序列操作,若没有 initializer 参数则抛出异常。
解决: 始终提供 initializer 参数:reduce(op, items, initial_value);或先检查序列是否为空。