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

无限迭代器

无限迭代器不会自动停止,必须配合 islicetakewhilebreak 使用,否则会进入死循环。

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)cachelru_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 装饰的函数传 listdict 参数时,报 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);或先检查序列是否为空。


参见

装饰器与函数高级
内置函数完全参考

阅读更多

Web 安全基础

1. HTML 转义(服务端渲染必须): 2. CSP(Content Security Policy): 3. HttpOnly Cookie:防止 JS 读取会话 Cookie: 4. 前端框架防护: 攻击者在第三方网站构造一个表单,诱导已登录用户提交,浏览器会自动携带目标站的 Cookie。 触发条件: 1. 用户已登录目标网站(Cookie 有效) 2. 目标 API 仅凭 Cookie 识别用户身份 3. 请求来源未验证 1. CSRF Token(推荐): 2. SameSite Cookie: 3. 验证 Origin/Referer 头:

By yellowdog

HTTP 协议深度指南

HTTP(HyperText Transfer Protocol)是 Web 的基础传输协议,基于 TCP/IP,采用请求/响应模型。 相关文档:Web安全基础(/web-an-quan-ji-chu/) FastAPI完全指南(/fastapi-wan-quan-zhi-nan/) Nginx完全指南(/nginx-wan-quan-zhi-nan/) 幂等性:多次执行相同请求,服务器状态结果相同。PUT /users/1 多次执行结果一致;POST /users 每次创建新资源,非幂等。 浏览器直接从本地缓存读取,不向服务器发送请求。 缓存命中时,状

By yellowdog

系统设计基础

SLA 对照表: 选择建议:无状态服务(Web 层、API 层)优先水平扩展;数据库初期垂直扩展,达到瓶颈后考虑分库分表或读写分离。 缓存穿透(查询不存在的 key,每次都打到 DB): 缓存击穿(热点 key 过期,瞬间大量请求打到 DB): 缓存雪崩(大量 key 同时过期,或缓存服务宕机): 令牌桶 Python 实现: Redis 实现分布式限流(滑动窗口): URL 命名规则: Cursor 分页响应格式: 雪花算法结构(64 bit): 定义:分布式系统不能同时满足以下三个特性: 在分布式环境中 P 是必须保证的,所以实际是 CP vs AP

By yellowdog

算法思路与模板

二分查找要求序列有序,每次将搜索范围缩减一半,时间复杂度 O(log n)。 两个指针从两端向中间收缩,常用于有序数组。 滑动窗口维护一个满足条件的区间 left, right,right 不断向右扩张,条件不满足时收缩 left。 滑动窗口通用框架: 1. 确定"子问题":原问题可以分解为哪些规模更小的同类问题 2. 定义 dpi 或 dpij 的含义,要足够清晰 3. 推导状态转移方程 4. 确定初始状态(边界条件) 5. 确定计算顺序(确保依赖的子问题先计算) 每件物品最多选一次。dpj = 容量为 j 时的最大价值,逆序遍历容量防止重复选取。 每

By yellowdog