Python 基础体系 · 第 44/112 篇。示例统一以 Python 3.14 为语言基线;第三方库使用与其兼容的现代稳定版本,版本敏感行为会单独说明。

Python functools:缓存、偏函数、归约、分派和装饰器工具

functools 是 Python 标准库中处理高阶函数和可调用对象的模块。所谓高阶函数,是指至少满足下列条件之一的函数:

  1. 接受函数作为参数;
  2. 返回一个函数或其他可调用对象;
  3. 改变可调用对象的行为或调用方式。

functools 的价值不在于把所有代码改写成“函数式风格”,而在于提供一组经过标准化的函数变换工具:缓存重复计算、固定部分参数、把序列归约成一个值、按类型选择实现、构造装饰器,以及兼容旧式比较函数。Python 3.14 还新增了 functools.Placeholder,使 partial() 可以预留任意位置的参数。(docs.python.org)


一、先建立统一模型:函数、可调用对象和函数变换

在 Python 中,“函数”不只包括使用 def 定义的对象。以下对象都可以被调用:

def add(x, y):
    return x + y


class Multiplier:
    def __call__(self, x, y):
        return x * y


multiply = Multiplier()

print(add(2, 3))       # 5
print(multiply(2, 3))  # 6

因此,functools 文档通常使用 callable,即“可调用对象”,而不是狭义的函数。一个高阶工具可能接收普通函数,也可能接收实现了 __call__() 的对象。

从抽象角度看,若函数:

f:A×BCf: A \times B \rightarrow C

表示它接收两个参数并返回一个结果,那么偏函数可以把其中一个参数固定下来,得到:

g:BCg: B \rightarrow C

并满足:

g(b)=f(a0,b)g(b) = f(a_0, b)

缓存则把函数扩展为带有内部映射的可调用对象:

F(x)={cache[x],x 已存在cache[x]f(x),x 不存在F(x) = \begin{cases} cache[x], & x \text{ 已存在}\\ cache[x] \leftarrow f(x), & x \text{ 不存在} \end{cases}

这两个过程都不是改变原函数的数学结果,而是改变调用协议执行路径


二、缓存:cachelru_cachecached_property

2.1 缓存成立的前提:相同输入必须允许复用相同输出

缓存适合具有以下性质的函数:

def f(x):
    ...

对于相同的 x,函数返回值应当稳定,或者至少在业务语义上允许复用。

更严格地说,如果缓存把调用参数 x 映射到结果 f(x),那么必须满足:

x1=x2f(x1)=f(x2)x_1 = x_2 \Rightarrow f(x_1) = f(x_2)

如果函数依赖当前时间、随机数、外部文件内容或数据库状态,这个条件可能不成立。

import functools
import time


@functools.cache
def current_time():
    return time.time()


first = current_time()
time.sleep(0.1)
second = current_time()

print(first == second)  # True

这里的结果是“错误的”还是“正确的”,取决于函数名称之外的业务语义。current_time() 的自然语义是每次获取当前时间,因此不应缓存;如果函数叫 startup_time(),并且明确表示进程启动时刻,那么缓存才可能合理。

官方文档明确指出,缓存不适合带副作用的函数、需要每次创建独立可变对象的函数、异步函数,以及 time()random() 这类不纯函数。(docs.python.org)


2.2 @cache:无上限的函数缓存

@functools.cache 是一个简单的无界缓存,效果等价于 @lru_cache(maxsize=None),但不需要维护淘汰顺序,因此实现更轻量。(docs.python.org)

from functools import cache


@cache
def factorial(n):
    print(f"calculate factorial({n})")
    return n * factorial(n - 1) if n else 1


print(factorial(5))
print(factorial(3))

第一次调用 factorial(5) 时,递归链为:

factorial(5)
  -> factorial(4)
    -> factorial(3)
      -> factorial(2)
        -> factorial(1)
          -> factorial(0)

随后缓存中存在:

0 -> 1
1 -> 1
2 -> 2
3 -> 6
4 -> 24
5 -> 120

再次调用 factorial(3) 时,函数体不会再次执行,而是直接读取缓存中的 3 -> 6

cache 适合输入空间有限、结果可长期复用的场景,例如递归动态规划、固定配置解析、不可变数据的昂贵转换。它不适合请求参数无限增长的长期运行进程,因为缓存不会自动删除旧条目。


2.3 @lru_cache:带容量限制的最近最少使用缓存

lru_cache 中的 LRU 是 Least Recently Used,即“最近最少使用”。当缓存达到 maxsize 后,新结果进入缓存,较久没有被访问的条目会被淘汰。

from functools import lru_cache


@lru_cache(maxsize=2)
def square(n):
    print(f"calculate square({n})")
    return n * n


square(1)  # miss,缓存:[1]
square(2)  # miss,缓存:[1, 2]
square(1)  # hit,1 变成最近使用,缓存逻辑顺序:[2, 1]
square(3)  # miss,淘汰 2,缓存:[1, 3]
square(2)  # miss,因为 2 已被淘汰

maxsize 越大,通常越容易命中,但会持有更多参数和返回值;maxsize=None 会关闭 LRU 淘汰,行为退化为无界缓存。缓存装饰器提供三个重要的诊断接口:

print(square.cache_info())
print(square.cache_parameters())

square.cache_clear()

典型输出形如:

CacheInfo(hits=1, misses=4, maxsize=2, currsize=2)
{'maxsize': 2, 'typed': False}

cache_info() 用于观察命中、未命中、容量和当前条目数;cache_parameters() 只是返回配置快照,修改这个字典不会改变缓存行为;cache_clear() 用于清空缓存。原始函数可以通过 __wrapped__ 访问。(docs.python.org)


2.4 缓存键:可哈希、参数形式和类型

lru_cachecache 使用函数参数构造缓存键,因此位置参数和关键字参数必须是可哈希对象。

from functools import lru_cache


@lru_cache
def total(values):
    return sum(values)


total([1, 2, 3])

运行时会得到:

TypeError: unhashable type: 'list'

可以把列表转换成元组:

total((1, 2, 3))  # 6

但“可哈希”只解决了缓存键能否建立的问题,不代表数据一定适合缓存。元组内部如果包含可变对象,仍然可能不可哈希;即使对象可哈希,如果对象状态参与结果却在缓存期间发生变化,也会破坏语义。

还要注意,参数的调用形式可能影响缓存条目。官方文档指出,下面两种调用可能被视为不同的缓存键:

f(a=1, b=2)
f(b=2, a=1)

因此,函数接口最好统一位置参数和关键字参数的使用方式,避免同一逻辑请求产生多个缓存条目。typed=True 时,不同类型的直接参数会分别缓存;typed=False 时通常会把部分等价值视为同一调用,但文档明确说明某些类型组合仍可能分开处理。(docs.python.org)


2.5 缓存并发不等于“只执行一次”

cachelru_cache 的内部缓存结构是线程安全的,多个线程并发更新时不会破坏缓存数据结构。但是,这不意味着同一个键的函数体绝对只执行一次。

时间线可能如下:

线程 A:检查 key,不存在
线程 A:开始计算

线程 B:检查 key,仍不存在
线程 B:也开始计算

线程 A:计算完成,写入结果
线程 B:计算完成,写入结果

因此,函数体仍可能被并发执行多次。若函数计算昂贵且要求“同一键只允许一个线程执行”,需要在业务层增加锁、请求合并或其他同步机制。标准缓存只保证缓存结构一致,不提供这种单飞语义。(docs.python.org)


2.6 方法缓存会持有 self

from functools import lru_cache


class Report:
    @lru_cache
    def render(self, page):
        return f"report={id(self)}, page={page}"

调用:

report = Report()
report.render(1)

缓存键实际上包含:

(self, 1)

只要缓存条目存在,self 实例就可能被缓存持有。对于大量短生命周期实例,这会延长对象生命周期并增加内存占用。官方文档明确说明,缓存方法时,self 会包含在缓存中。(docs.python.org)

如果缓存内容属于实例自身,常见选择是让缓存键使用稳定的不可变字段,或者把缓存放在实例中并明确管理生命周期,而不是无条件给实例方法加上全局式装饰器。


2.7 cached_property:把一次计算变成实例属性

cached_property 适用于“第一次访问时计算,之后像普通实例属性一样读取”的场景。

from functools import cached_property
import statistics


class DataSet:
    def __init__(self, values):
        self._values = tuple(values)

    @cached_property
    def stdev(self):
        print("计算标准差")
        return statistics.stdev(self._values)


data = DataSet([10, 20, 30])

print(data.stdev)  # 计算标准差,然后输出 10.0
print(data.stdev)  # 直接读取实例属性,不再计算

它的状态变化可以表示为:

实例创建:
data.__dict__ = {"_values": (10, 20, 30)}

第一次读取 data.stdev:
1. 实例中没有 "stdev"
2. 执行方法
3. 将结果写入 data.__dict__["stdev"]

之后读取 data.stdev:
1. 实例中已有 "stdev"
2. 直接返回普通属性值

这与 property 的关键区别是:普通 property 通常拦截读取,并阻止没有 setter 的属性写入;cached_property 第一次计算后会写入实例属性,之后用户甚至可以覆盖它。删除属性后,可以触发下一次重新计算:

del data.stdev
print(data.stdev)  # 再次计算

cached_property 要求实例具有可变的 __dict__,因此不适用于没有 __dict____slots__ 实例,也会影响 key-sharing dictionary 的空间特性。Python 3.12 起,标准实现移除了此前存在的、并且是“每个属性共享一个锁”的未公开锁;在 Python 3.14 中,多线程访问同一实例时 getter 仍可能执行多次,需要自行同步。(docs.python.org)


三、偏函数:partialpartialmethodPlaceholder

3.1 partial:固定一部分参数

functools.partial() 接收一个可调用对象和一部分参数,返回一个新的可调用对象。

from functools import 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 square(base):
    return power(base, exponent=2)

partial 不需要重新定义函数体,并且保留了底层对象、固定位置参数和固定关键字参数:

print(square.func is power)       # True
print(square.args)               # ()
print(square.keywords)           # {'exponent': 2}

如果固定的是位置参数:

def format_name(first, last, separator=", "):
    return f"{last}{separator}{first}"


format_chinese = partial(format_name, separator=" ")
print(format_chinese("三", "张"))  # 张 三

调用时新增的关键字参数会扩展并覆盖固定关键字参数,新增的位置参数会追加到已经固定的位置参数之后。(docs.python.org)


3.2 Python 3.14 的 Placeholder:预留任意位置

没有 Placeholder 时,partial() 主要固定左侧的位置参数:

def join3(a, b, c):
    return f"{a}-{b}-{c}"


f = partial(join3, "A")
print(f("B", "C"))  # A-B-C

如果希望固定中间参数,就需要使用 Python 3.14 新增的 functools.Placeholder

from functools import partial, Placeholder


middle_fixed = partial(join3, Placeholder, "B", Placeholder)

print(middle_fixed("A", "C"))  # A-B-C

调用时,传入的位置参数会按顺序填充占位符:

固定模板:Placeholder, "B", Placeholder
调用参数:"A", "C"

填充结果:"A", "B", "C"

所有占位符都必须在调用时被填充,否则会抛出 TypeError

middle_fixed("A")
# TypeError

占位符只能用于位置参数,不能作为关键字参数值传入。它还可以和已有的 partial 继续组合:

from functools import partial, Placeholder as _


remove = partial(str.replace, _, _, "")
print(remove("Hello, dear world!", " dear"))
# Hello, world!

这里第一个 _ 表示待处理字符串,第二个 _ 表示要替换的内容。Placeholder 是 Python 3.14 的版本敏感能力,使用前应确认运行环境确实是 Python 3.14 或更高版本。(docs.python.org)


3.3 partialmethod:在类定义中固定方法参数

partialmethodpartial 的区别不在于“是否固定参数”,而在于描述器行为

from functools import partialmethod


class Cell:
    def __init__(self):
        self.alive = False

    def set_state(self, state):
        self.alive = bool(state)

    set_alive = partialmethod(set_state, True)
    set_dead = partialmethod(set_state, False)


cell = Cell()
cell.set_alive()
print(cell.alive)  # True

cell.set_dead()
print(cell.alive)  # False

调用 cell.set_alive() 时,实际参数顺序近似为:

Cell.set_state(cell, True)

也就是说,绑定实例的 self 会先插入,然后才是 partialmethod 固定的参数。partialmethod 适合把一个通用方法暴露成多个语义更明确的方法,例如:

class Request:
    def set_status(self, status):
        self.status = status

    mark_success = partialmethod(set_status, 200)
    mark_not_found = partialmethod(set_status, 404)

它依赖 Python 的描述器协议:从实例访问方法时,类属性会根据实例生成绑定方法。普通 partial 本身不自动承担这种方法绑定语义,而 partialmethod 专门为类属性中的方法定义设计。(docs.python.org)


四、归约:reduce 如何把多个值折叠成一个值

4.1 左折叠的形式化定义

functools.reduce(function, iterable, initial) 从左到右累计调用二元函数。

对于序列:

[x1, x2, x3, x4]

没有初始值时,计算过程为:

(((x1x2)x3)x4)(((x_1 \mathbin{\circ} x_2) \mathbin{\circ} x_3) \mathbin{\circ} x_4)

若初始值为 zz,则过程为:

(((zx1)x2)x3)x4(((z \mathbin{\circ} x_1) \mathbin{\circ} x_2) \mathbin{\circ} x_3) \mathbin{\circ} x_4

其中:

  • function 是二元函数;
  • 左参数是当前累计值;
  • 右参数是迭代器取出的下一个元素;
  • 最终结果只有一个值。
from functools import reduce
import operator


result = reduce(operator.mul, [2, 3, 4], 1)
print(result)  # 24

中间状态是:

初始值 = 1
1 * 2 = 2
2 * 3 = 6
6 * 4 = 24

等价循环为:

result = 1
for value in [2, 3, 4]:
    result = operator.mul(result, value)

官方文档给出的近似实现也体现了这一点:没有 initial 时先取第一个元素作为累计值;有 initial 时直接以它开始。Python 3.14 起,initial 还可以作为关键字参数传入。(docs.python.org)


4.2 空序列为什么需要 initial

reduce(operator.add, [])

会抛出:

TypeError: reduce() of empty iterable with no initial value

原因是:没有初始值时,reduce 需要先取第一个元素作为累计值,但空迭代器没有第一个元素。

传入初始值后:

print(reduce(operator.add, [], 0))  # 0
print(reduce(operator.mul, [], 1))  # 1

这里的 01 分别是加法和乘法的单位元:

0+x=x0 + x = x

1×x=x1 \times x = x

如果归约操作具有单位元,使用 initial 通常能让空输入也有明确结果。若没有自然单位元,就必须决定空输入是异常、特殊对象还是业务默认值。


4.3 结合律决定归约能否安全重排

如果操作满足结合律:

(ab)c=a(bc)(a \circ b) \circ c = a \circ (b \circ c)

那么不同分组方式通常不会影响结果。例如整数加法和乘法满足结合律。

但浮点加法不严格满足结合律:

from functools import reduce


values = [1e16, 1.0, -1e16]

left = reduce(lambda x, y: x + y, values)
right = values[0] + (values[1] + values[2])

print(left)   # 0.0
print(right)  # 1.0

因此,reduce 的结果不仅与元素和操作有关,还与顺序、分组方式和数值类型有关。对于字符串拼接、列表连接等操作,reduce 还可能产生大量中间对象:

reduce(lambda a, b: a + b, many_strings)

如果只是求和,应优先使用 sum();如果需要保留每一步中间结果,应使用 itertools.accumulate()reduce 本身只返回最终值,并且不能用于无限迭代器,因为它必须等到输入耗尽后才能返回最终结果。(docs.python.org)


五、分派:singledispatchsingledispatchmethod

5.1 单分派不是多参数重载

singledispatch 将一个函数转换成泛型函数,根据第一个参数的运行时类型选择实现。

from functools import singledispatch


@singledispatch
def describe(value):
    return f"默认实现:{value!r}"


@describe.register
def _(value: int):
    return f"整数:{value}"


@describe.register
def _(value: list):
    return f"列表,长度:{len(value)}"


print(describe(10))       # 整数:10
print(describe([1, 2]))   # 列表,长度:2
print(describe("hello"))  # 默认实现:'hello'

这里的 describe 不是根据参数数量、参数名称或多个参数的类型组合进行分派,而是只查看第一个参数。

注册过程可以写成显式类型形式:

@describe.register(int)
def describe_int(value):
    return f"整数:{value}"

也可以使用类型注解自动推断:

@describe.register
def _(value: float):
    return f"浮点数:{value}"

如果没有更具体的实现,原始函数作为 object 类型的默认实现。注册抽象基类时,其虚拟子类也可以匹配;选择具体实现时,会依据类型层次和方法解析顺序寻找最合适的实现。(docs.python.org)


5.2 分派表、诊断和错误路径

泛型函数提供 dispatch() 和只读的 registry

print(describe.dispatch(int))
print(describe.dispatch(dict))
print(describe.registry.keys())

其中:

  • dispatch(int) 返回处理 int 的函数;
  • dispatch(dict) 如果没有注册 dict,会返回默认实现;
  • registry 展示已经注册的类型和实现。

默认实现不一定应该“默默处理所有类型”。如果某个类型没有业务意义,可以让默认函数抛出异常:

@singledispatch
def serialize(value):
    raise TypeError(f"不支持的类型:{type(value).__name__}")

这样,类型遗漏会在调用点暴露,而不是被转换成错误但看似合法的数据。


5.3 singledispatchmethod 的分派参数

在方法中使用 singledispatchmethod 时,分派对象不是 selfcls,而是第一个非 self、非 cls 参数。

from functools import singledispatchmethod


class Negator:
    @singledispatchmethod
    def negate(self, value):
        raise TypeError(f"不支持:{type(value).__name__}")

    @negate.register
    def _(self, value: int):
        return -value

    @negate.register
    def _(self, value: bool):
        return not value


negator = Negator()

print(negator.negate(3))      # -3
print(negator.negate(True))   # False

类方法、静态方法等装饰器与 singledispatchmethod 嵌套时,singledispatchmethod 必须放在最外层,否则可能无法通过 dispatcher.register 注册实现:

class Factory:
    @singledispatchmethod
    @classmethod
    def build(cls, value):
        raise TypeError(type(value).__name__)

    @build.register
    @classmethod
    def _(cls, value: int):
        return cls(value)

这个顺序反映了装饰器的执行规则:靠近 def 的装饰器先应用,外层装饰器最后接收结果。(docs.python.org)


六、装饰器工具:update_wrapperwraps__wrapped__

6.1 装饰器本质上是函数替换

下面的语法:

@trace
def add(x, y):
    return x + y

等价于:

def add(x, y):
    return x + y

add = trace(add)

因此,装饰器必须返回一个新的可调用对象,或者返回原对象本身。

一个最小装饰器如下:

def trace(func):
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        return func(*args, **kwargs)

    return wrapper

但这样会丢失元数据:

@trace
def add(x: int, y: int) -> int:
    """计算两个整数的和。"""
    return x + y


print(add.__name__)  # wrapper
print(add.__doc__)   # None

函数对象的名称、限定名、文档字符串和注解可能被包装函数覆盖。


6.2 wrapsupdate_wrapper 的便捷形式

正确写法是:

from functools import wraps


def trace(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        return func(*args, **kwargs)

    return wrapper


@trace
def add(x: int, y: int) -> int:
    """计算两个整数的和。"""
    return x + y


print(add.__name__)  # add
print(add.__doc__)   # 计算两个整数的和。
print(add.__annotations__)  # {'x': <class 'int'>, ...}

wraps(func) 等价于:

partial(update_wrapper, wrapped=func)

它会把被包装函数的元数据复制或更新到包装函数,并设置:

wrapper.__wrapped__ = func

__wrapped__ 使 inspect 等工具能够继续找到原始函数,也可以绕过缓存或装饰器直接访问底层实现:

original_add = add.__wrapped__
print(original_add(2, 3))  # 5

默认情况下,update_wrapper() 会处理 __module____name____qualname____annotations____type_params____doc__,并更新包装函数的 __dict__。Python 3.12 起,__type_params__ 也在默认复制范围内。(docs.python.org)


6.3 参数化装饰器:三层调用结构

参数化装饰器比普通装饰器多一层函数:

from functools import wraps


def retry(times):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None

            for attempt in range(times):
                try:
                    return func(*args, **kwargs)
                except Exception as exc:
                    last_error = exc

            raise last_error

        return wrapper

    return decorator

使用时:

@retry(times=3)
def read_config():
    ...

等价于:

read_config = retry(times=3)(read_config)

调用过程是:

retry(times=3)
    -> 返回 decorator

decorator(read_config)
    -> 返回 wrapper

wrapper()
    -> 最多调用 read_config 三次

错误处理必须明确:上例只在函数抛出异常时重试,三次失败后重新抛出最后一个异常。如果操作不是幂等的,例如扣款、创建订单、发送消息,盲目重试可能造成重复副作用。装饰器只负责控制调用流程,不能自动证明重试是安全的。


七、比较适配:cmp_to_key

Python 的排序 API 使用 key 函数。key(x) 接收一个元素并返回排序依据:

names = ["bob", "Alice", "carol"]
print(sorted(names, key=str.lower))
# ['Alice', 'bob', 'carol']

旧式比较函数则接收两个元素,并返回:

  • 负数:第一个元素更小;
  • 零:两者相等;
  • 正数:第一个元素更大。

cmp_to_key() 将旧式比较函数转换为现代排序 API 所需的 key 函数:

from functools import cmp_to_key


def compare_length(left, right):
    if len(left) < len(right):
        return -1
    if len(left) > len(right):
        return 1
    return (left > right) - (left < right)


values = ["bbb", "a", "cc", "ab"]
print(sorted(values, key=cmp_to_key(compare_length)))
# ['a', 'ab', 'cc', 'bbb']

它不是让 sorted() 重新接受比较函数,而是构造一个包装对象,使排序算法可以通过对象之间的比较完成旧式规则。对于新代码,如果排序规则可以直接表达为 key,直接使用 key 函数通常更清晰;cmp_to_key 主要用于兼容已有比较函数,例如本地化字符串比较。(docs.python.org)


八、total_ordering:自动补齐比较方法

total_ordering 是类装饰器。类只实现一个排序方法和 __eq__(),它会补齐其他富比较方法。

from functools import total_ordering


@total_ordering
class Version:
    def __init__(self, major, minor):
        self.major = major
        self.minor = minor

    def _key(self):
        return self.major, self.minor

    def __eq__(self, other):
        if not isinstance(other, Version):
            return NotImplemented
        return self._key() == other._key()

    def __lt__(self, other):
        if not isinstance(other, Version):
            return NotImplemented
        return self._key() < other._key()


v1 = Version(3, 10)
v2 = Version(3, 14)

print(v1 < v2)   # True
print(v1 <= v2)  # True
print(v1 > v2)   # False
print(v1 != v2)  # True

这里的 NotImplementedFalse 不同。NotImplemented 表示当前比较方法无法处理对方类型,Python 可以尝试反向比较或最终产生适当结果;直接返回 False 则表示比较已经有了确定结论。

total_ordering 的代价是自动生成的方法会增加调用层次,执行速度和 traceback 复杂度可能不如手写全部比较方法。如果比较操作位于性能关键路径,应该通过基准测试确认是否需要显式实现全部方法。该装饰器也不会覆盖类本身或父类已经声明的方法。(docs.python.org)


九、一个组合示例:按类型分派、缓存和元数据保留

下面将多个工具组合起来,构造一个简单的序列化入口:

from functools import singledispatch, lru_cache, wraps


def audit(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[audit] {func.__name__}")
        return func(*args, **kwargs)

    return wrapper


@lru_cache(maxsize=128)
def normalize_text(value):
    print("执行文本规范化")
    return value.strip().lower()


@singledispatch
@audit
def encode(value):
    raise TypeError(f"不支持的类型:{type(value).__name__}")


@encode.register
def _(value: str):
    return normalize_text(value).encode("utf-8")


@encode.register
def _(value: bytes):
    return value


print(encode(" Hello "))
print(encode(" Hello "))
print(encode(b"raw"))

print(encode.dispatch(str).__name__)
print(normalize_text.cache_info())

执行逻辑如下:

  1. encode 根据第一个参数的类型进行分派;
  2. 字符串进入 str 实现;
  3. normalize_text 第一次处理 " Hello "
  4. 第二次相同调用命中 LRU 缓存;
  5. bytes 直接返回;
  6. @audit 通过 @wraps 保留默认实现的元数据;
  7. dispatch()cache_info() 用于检查实际选择和缓存效果。

这里有一个装饰器顺序要点:

@singledispatch
@audit
def encode(value):
    ...

等价于:

encode = singledispatch(audit(encode))

如果把顺序调换,注册接口和被包装对象的关系会改变。涉及 singledispatchclassmethodstaticmethod 或自定义装饰器时,应先展开成赋值形式检查装饰顺序,再决定代码结构。


十、工具之间的边界

cachelru_cache

  • 结果可以长期保留、键空间有限:使用 cache
  • 进程长期运行、需要控制内存:使用 lru_cache(maxsize=...)
  • 结果依赖时间、随机数、外部状态或副作用:通常不要使用;
  • 方法缓存:确认 self 被持有的生命周期和内存影响。

cached_propertylru_cache

  • cached_property 把结果写入实例属性,支持实例级覆盖和删除;
  • lru_cache 把调用参数放入函数级缓存,方法调用时通常包含 self
  • cached_property 依赖实例 __dict__
  • 多线程下,cached_property getter 可能执行多次,需要 getter 幂等或自行加锁。

partial 与 lambda

以下两种写法都能固定参数:

from functools import partial

double_a = partial(pow, 2)
double_b = lambda x: pow(2, x)

partial 更明确地表达了“已有函数固定部分参数”,并保留了 funcargskeywords 等结构化信息;lambda 则适合需要额外逻辑的情况。

reduce 与循环

如果归约步骤需要复杂条件、异常分支、日志或多个状态变量,普通 for 循环往往更容易验证。reduce 最适合操作简单、累计状态清晰、左折叠含义明确的场景。

singledispatch 与显式 if

类型分派规则少且固定时,if isinstance(...) 可能更直接;当实现需要按模块注册、允许扩展或类型分支逐渐增长时,singledispatch 可以把“选择逻辑”和“具体实现”分离。

wraps 与行为透明

wraps 只复制元数据,不会自动复制函数签名的真实调用逻辑,也不会让包装器自动具备原函数的类型检查、资源管理或异常语义。包装器仍然必须正确转发:

return func(*args, **kwargs)

如果包装器改变了参数、返回值或异常,就应在文档和类型注解中明确反映这种变化,而不能因为使用了 @wraps 就假装行为完全没有改变。


functools 的核心可以归纳为五种函数变换:

缓存:把重复调用变成结果查找
偏函数:把完整函数变成参数更少的新调用入口
归约:把多个输入折叠成一个累计结果
分派:把一个入口映射到多个类型实现
装饰:在保留调用接口的前提下包裹执行过程

它们都建立在同一个基础上:Python 中函数是对象,可被保存、传递、组合和再次包装。真正重要的不是记住每个 API 的名称,而是判断一次变换改变了什么状态、保留了什么语义,以及在哪些输入、并发和生命周期条件下,这种改变仍然成立。


系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理;正文、示例与工程清单由 WR BLOG 编写。