Python functools 深度解析：lru_cache、wraps 与 partial 的底层逻辑

1
# 简化的 LRU 缓存实现（理解原理用）
2
from functools import wraps
3
from collections import OrderedDict
4

5
def simple_lru_cache(maxsize: int = 128):
6
    def decorator(func):
7
        cache = OrderedDict()
8

9
        @wraps(func)
10
        def wrapper(*args, **kwargs):
11
            key = (args, tuple(sorted(kwargs.items())))
12

13
            if key in cache:
14
                # 命中：移到末尾（标记为最近使用）
15
                cache.move_to_end(key)
16
                return cache[key]
17

18
            # 未命中：计算并存储
19
            result = func(*args, **kwargs)
20
            cache[key] = result
21

22
            # 如果超出容量，弹出最久未使用的（头部）
23
            if len(cache) > maxsize:
24
                cache.popitem(last=False)
25

26
            return result
27

28
        # 暴露缓存统计信息
29
        wrapper.cache_info = lambda: (
30
            f"hits=0, misses=0, maxsize={maxsize}, currsize={len(cache)}"
31
        )
32
        wrapper.cache_clear = cache.clear
33
        return wrapper
34
    return decorator

1
from functools import lru_cache
2

3
@lru_cache(maxsize=3)
4
def expensive_compute(x: int) -> int:
5
    print(f"  计算 {x}...")
6
    return x * x
7

8
print(expensive_compute(1))  # 计算 1... → 1
9
print(expensive_compute(2))  # 计算 2... → 4
10
print(expensive_compute(3))  # 计算 3... → 9
11
print(expensive_compute(4))  # 计算 4... → 16（缓存满了，淘汰 1）
12
print(expensive_compute(1))  # 计算 1... → 1（已被淘汰，重新计算）
13

14
print(expensive_compute.cache_info())
15
# CacheInfo(hits=0, misses=5, maxsize=3, currsize=3)

1
WRAPPER_ASSIGNMENTS = ('__module__', '__name__', '__qualname__',
2
                       '__annotations__', '__doc__')
3
WRAPPER_UPDATES = ('__dict__',)
4

5
def my_update_wrapper(wrapper, wrapped):
6
    # 1. 拷贝属性
7
    for attr in WRAPPER_ASSIGNMENTS:
8
        try:
9
            value = getattr(wrapped, attr)
10
        except AttributeError:
11
            pass
12
        else:
13
            setattr(wrapper, attr, value)
14

15
    # 2. 更新字典（保留 wrapper 自己的 __dict__，同时合并 wrapped 的）
16
    for attr in WRAPPER_UPDATES:
17
        getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
18

19
    # 3. 添加 __wrapped__ 属性，保留对原始函数的引用
20
    wrapper.__wrapped__ = wrapped
21

22
    return wrapper

1
from functools import wraps
2
import inspect
3

4
def decorator_a(func):
5
    @wraps(func)
6
    def wrapper_a(*args, **kwargs):
7
        return func(*args, **kwargs)
8
    return wrapper_a
9

10
def decorator_b(func):
11
    @wraps(func)
12
    def wrapper_b(*args, **kwargs):
13
        return func(*args, **kwargs)
14
    return wrapper_b
15

16
@decorator_a
17
@decorator_b
18
def target(x: int) -> int:
19
    """目标函数"""
20
    return x * 2
21

22
# 直接看签名
23
print(inspect.signature(target))        # (x: int) -> int ✅
24
print(target.__wrapped__.__name__)      # wrapper_b（decorator_b 的 wrapper）
25

26
# unwrap 追溯最原始函数
27
original = inspect.unwrap(target)
28
print(original.__name__)  # target ✅

1
import tkinter as tk
2
from functools import partial
3

4
def on_button_click(button_id: int):
5
    print(f"按钮 {button_id} 被点击")
6

7
root = tk.Tk()
8
for i in range(1, 6):
9
    # ❌ 错误：lambda 会捕获循环变量 i 的最终值
10
    # btn = tk.Button(root, text=f"按钮{i}", command=lambda: on_button_click(i))
11

12
    # ✅ 正确：partial 在创建时就固定了参数
13
    btn = tk.Button(root, text=f"按钮{i}", command=partial(on_button_click, i))
14
    btn.pack()
15

16
root.mainloop()

1
from functools import partialmethod
2

3
class Cell:
4
    def __init__(self):
5
        self.alive = False
6

7
    def set_state(self, state: str):
8
        self.alive = (state == "ALIVE")
9

10
    # 预设两个便捷方法
11
    set_alive = partialmethod(set_state, "ALIVE")
12
    set_dead = partialmethod(set_state, "DEAD")
13

14
cell = Cell()
15
cell.set_alive()
16
print(cell.alive)  # True
17
cell.set_dead()
18
print(cell.alive)  # False

1
from functools import singledispatch
2

3
@singledispatch
4
def process(data):
5
    """默认实现"""
6
    raise NotImplementedError(f"不支持的类型: {type(data)}")
7

8
@process.register
9
def _(data: str):
10
    print(f"处理字符串: {data.upper()}")
11
    return len(data)
12

13
@process.register
14
def _(data: int):
15
    print(f"处理整数: {data * 2}")
16
    return data * 2
17

18
@process.register
19
def _(data: list):
20
    print(f"处理列表: 共 {len(data)} 个元素")
21
    return sum(data)
22

23
process("hello")       # 处理字符串: HELLO → 5
24
process(42)            # 处理整数: 84 → 84
25
process([1, 2, 3])     # 处理列表: 共 3 个元素 → 6
26
process(3.14)          # NotImplementedError: 不支持的类型: <class 'float'>

1
# 简化的 singledispatch 实现
2
class SimpleDispatcher:
3
    def __init__(self, default_func):
4
        self.default = default_func
5
        self.registry = {}
6

7
    def register(self, type_hint, func=None):
8
        if func is None:
9
            # @dispatch.register 作为装饰器使用
10
            def decorator(f):
11
                self.registry[type_hint] = f
12
                return f
13
            return decorator
14
        # @dispatch.register(SomeType, func) 直接注册
15
        self.registry[type_hint] = func
16

17
    def __call__(self, arg, *args, **kwargs):
18
        # 查找 arg 类型对应的处理函数
19
        arg_type = type(arg)
20
        handler = self.registry.get(arg_type, self.default)
21
        return handler(arg, *args, **kwargs)

1
from functools import singledispatch
2
from datetime import datetime
3
from enum import Enum
4

5
class Status(Enum):
6
    OK = "ok"
7
    FAIL = "fail"
8

9
@singledispatch
10
def serialize(obj):
11
    return str(obj)
12

13
@serialize.register
14
def _(obj: datetime):
15
    return obj.isoformat()
16

17
@serialize.register
18
def _(obj: Enum):
19
    return obj.value
20

21
@serialize.register(list)
22
def _(obj: list):
23
    return [serialize(item) for item in obj]
24

25
@serialize.register(dict)
26
def _(obj: dict):
27
    return {k: serialize(v) for k, v in obj.items()}
28

29
# 测试
30
data = {
31
    "name": "测试",
32
    "time": datetime(2026, 5, 30, 12, 0),
33
    "status": Status.OK,
34
    "items": [1, datetime(2026, 1, 1), Status.FAIL],
35
}
36
print(serialize(data))
37
# {'name': '测试', 'time': '2026-05-30T12:00:00', 'status': 'ok', 'items': [1, '2026-01-01T00:00:00', 'fail']}

1
def get_nested(data: dict, path: str, default=None):
2
    """通过 'a.b.c' 路径访问嵌套字典"""
3
    keys = path.split(".")
4
    try:
5
        return reduce(lambda d, key: d[key], keys, data)
6
    except (KeyError, TypeError):
7
        return default
8

9
config = {
10
    "server": {
11
        "database": {
12
            "host": "localhost",
13
            "port": 5432,
14
        }
15
    }
16
}
17

18
print(get_nested(config, "server.database.host"))  # localhost
19
print(get_nested(config, "server.database.port"))  # 5432
20
print(get_nested(config, "server.missing.key"))    # None

1
class CmpKey:
2
    def __init__(self, obj, cmp):
3
        self.obj = obj
4
        self.cmp = cmp
5

6
    def __lt__(self, other):
7
        return self.cmp(self.obj, other.obj) < 0
8

9
    def __gt__(self, other):
10
        return self.cmp(self.obj, other.obj) > 0
11

12
    def __eq__(self, other):
13
        return self.cmp(self.obj, other.obj) == 0
14

15
    def __le__(self, other):
16
        return self.cmp(self.obj, other.obj) <= 0
17

18
    def __ge__(self, other):
19
        return self.cmp(self.obj, other.obj) >= 0

1
from functools import cached_property
2
import time
3

4
class DataSet:
5
    def __init__(self, data: list[int]):
6
        self.data = data
7

8
    @cached_property
9
    def mean(self) -> float:
10
        print("  计算平均值...")
11
        time.sleep(1)  # 模拟耗时计算
12
        return sum(self.data) / len(self.data)
13

14
    @cached_property
15
    def std_dev(self) -> float:
16
        print("  计算标准差...")
17
        time.sleep(1)
18
        m = self.mean  # 不会重新计算！
19
        return (sum((x - m) ** 2 for x in self.data) / len(self.data)) ** 0.5
20

21
ds = DataSet([1, 2, 3, 4, 5])
22

23
# 第一次访问
24
print(ds.mean)      # 计算平均值... → 3.0
25
print(ds.std_dev)   # 计算标准差... → 1.414...
26

27
# 第二次访问（命中缓存，不重新计算）
28
print(ds.mean)      # → 3.0（没有打印"计算平均值..."）