RSS
菜单
全部文章快讯开发科技深度热点

装饰器:从函数到框架(Python 从精通到入门 · 03)

内容摘要

装饰器是「接收函数、返回函数」的普通函数。就这一句话,撑起了缓存、重试、鉴权、路由——以及大半个框架世界。

装饰器是「接收函数、返回函数」的普通函数。就这一句话,撑起了缓存、重试、鉴权、路由——以及大半个框架世界。

你将学到

  • 闭包回顾:为什么装饰器能记住外层变量
  • 不带参 / 带参装饰器的写法,以及 functools.wraps 到底修了什么
  • 类装饰器、装饰器叠加顺序,以及装饰器与描述符的组合
  • 用装饰器实现缓存(lru_cache 原理)、重试、计时、权限校验

前置知识

02 - 描述符协议。后文代码默认已 import functools, time。


一、函数是对象,闭包是它的外挂

函数可以赋值、传参、返回;返回的内层函数会记住外层作用域的变量,这就是闭包:

def make_counter():
    count = 0
    def counter():
        nonlocal count            # 声明 count 来自外层,而不是新建局部变量
        count += 1
        return count
    return counter

c = make_counter()
print(c(), c(), c())                          # 输出: 1 2 3

make_counter() 早已返回,但 count 还活着。装饰器要记的所有状态都存这里:没有闭包,装饰器只能「调用前后各打一行」;有了闭包,才能「调用 N 次」「缓存 N 个结果」。

二、最朴素的装饰器

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

@log
def add(a, b):
    return a + b

print(add(1, 2))        # 输出: 先打印「调用 add(1, 2)」,再打印 3
print(add.__name__)     # 输出: wrapper —— 元信息被吃掉了(见下一节)

@log 只是语法糖:add = log(add)。名字 add 现在指向 wrapper,而 wrapper 从闭包里捕获了原来的 add。同一个装饰器装饰两个函数,会生成两个独立闭包,互不干扰。

三、functools.wraps:别把元信息弄丢

上面的 add.__name__ 变成了 wrapper、__doc__ 也没了。这会坑到 inspect.signature、文档生成、pytest 的用例发现,以及 FastAPI 这种靠签名做依赖注入的框架:

def log(func):
    @functools.wraps(func)          # 复制 __name__ / __doc__ / __module__ / __qualname__ ...
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@log
def add(a, b):
    """两数相加"""
    return a + b

print(add.__name__, add.__doc__)    # 输出: add 两数相加

wraps 顺手留下的 __wrapped__,是 inspect 用来「穿透装饰器看真实签名」的钩子。写装饰器的第一反应就该是加上它。

四、带参数的装饰器:多包一层

想写 @repeat(3),就得再套一层函数——因为此时 repeat(3) 先执行、返回真正的装饰器:

def repeat(times):
    def decorator(func):                    # 这一层才是装饰器
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return [func(*args, **kwargs) for _ in range(times)]
        return wrapper
    return decorator

@repeat(3)
def ping():
    return "pong"

print(ping())             # 输出: ['pong', 'pong', 'pong']

等价于 ping = repeat(3)(ping)。记忆方法:不带参两层,带参三层——层数就是「谁先被调用」。

五、类装饰器

任何可调用对象都能当装饰器,类当然可以。装成实例后能自然携带状态:

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0
        functools.update_wrapper(self, func)     # 类实例版的 wraps

    def __call__(self, *args, **kwargs):
        self.count += 1
        return self.func(*args, **kwargs)
@CountCalls
def say(name):
    return f"hi {name}"

print(say("Ada"))      # 输出: hi Ada
print(say.count)       # 输出: 1(状态天然存在实例上)

装饰器自身要带参时,写成 class CountCalls: def __init__(self, times): ... 即可:先构造实例、再由实例去装饰,形态变了,原理不变。

六、叠加顺序:就近包装,外层先跑

def tag(label):
    def deco(func):
        @functools.wraps(func)
        def wrapper(*a, **k):
            print(f"-> {label}")
            r = func(*a, **k)
            print(f"<- {label}")
            return r
        return wrapper
    return deco
@tag("A")
@tag("B")
def hello():
    print("hello")

hello()        # 输出: -> A / -> B / hello / <- B / <- A(每行一次)

两条规则:靠近 def 的先被包装(hello = tag("A")(tag("B")(hello)),B 在里、A 在外);外层先进入、后退出(A 的 -> 最先打印、<- 最后打印)。所以「鉴权、计时、日志放最外,缓存藏最里」通常是对的。

七、实战:缓存——lru_cache 的原理

缓存的全部秘密,就是闭包里的一个 dict:

def memoize(func):
    cache = {}                              # 藏在闭包里,每次调用都能读到
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)       # 只有第一次真正计算
        return cache[args]
    return wrapper

@memoize
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

print(fib(30))        # 输出: 832040(不加缓存要算上百万次,这里几乎瞬时)

标准库版本多了「容量上限 + 最近最少使用淘汰 + 线程安全 + 统计」:

@functools.lru_cache(maxsize=None)          # maxsize=None 表示不淘汰
def fib2(n):
    return n if n < 2 else fib2(n - 1) + fib2(n - 2)

print(fib2(100))        # 输出: 354224848179261915075
# fib2.cache_info() -> CacheInfo(hits=98, misses=101, maxsize=None, currsize=101)

限制也清楚:参数必须可哈希(args 里不能有 list)。Python 3.8+ 的 functools.cache 就是 lru_cache(maxsize=None) 的简写。

八、实战:重试、计时、权限校验

# 1) 计时:finally 保证异常路径也打印
def timing(func):
    @functools.wraps(func)
    def wrapper(*a, **k):
        t0 = time.perf_counter()
        try:
            return func(*a, **k)
        finally:
            print(f"{func.__name__} 耗时 {time.perf_counter() - t0:.4f}s")
    return wrapper

# 2) 重试:带参三层;最后一次失败必须把异常抛出去,不能吞
def retry(times=3, delay=0.01, exc=Exception):
    def deco(func):
        @functools.wraps(func)
        def wrapper(*a, **k):
            for attempt in range(1, times + 1):
                try:
                    return func(*a, **k)
                except exc:
                    if attempt == times:
                        raise
                    time.sleep(delay)
        return wrapper
    return deco

_calls = 0
@retry(times=3)
def flaky():
    global _calls
    _calls += 1
    if _calls < 3:
        raise ConnectionError("网络抖动")
    return f"第 {_calls} 次成功"

print(flaky())      # 输出: 第 3 次成功(前两次静默重试)
# 3) 权限校验:框架里最典型的一类
def require_role(role):
    def deco(func):
        @functools.wraps(func)
        def wrapper(self, *a, **k):
            if role not in getattr(self, "roles", ()):
                raise PermissionError(f"需要角色 {role}")
            return func(self, *a, **k)
        return wrapper
    return deco

class User:
    def __init__(self, roles):
        self.roles = roles
    @require_role("admin")
    def delete_all(self):
        return "已清空"
print(User(["admin"]).delete_all())   # 输出: 已清空
User(["guest"]).delete_all()          # PermissionError: 需要角色 admin

九、装饰器 + 描述符:框架里的组合拳

装饰器返回普通函数时 self 会自动绑定;若返回的是类实例(为了携带复杂状态),就必须实现 __get__,否则 d.heavy() 会把 self 传错:

class Cached:
    """把结果缓存在实例上的装饰器(类实现,所以必须实现 __get__)"""
    def __init__(self, func):
        self.func = func
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return functools.partial(self.__call__, obj)    # 绑定实例
    def __call__(self, obj, *args, **kwargs):
        if "_cached" not in obj.__dict__:
            obj.__dict__["_cached"] = self.func(obj, *args, **kwargs)
        return obj.__dict__["_cached"]

class Data:
    @Cached
    def heavy(self):
        print("(真正计算了一次)")
        return 42
d = Data()
print(d.heavy())     # 输出: (真正计算了一次) 然后 42
print(d.heavy())     # 输出: 42(命中缓存)

为什么要 __get__?Cached 实例存在类字典里,属性查找会走描述符协议(见 02 - 描述符协议)。少了它,d.heavy 直接返回 Cached 实例,调用时缺 obj 就崩。Django 的 @cached_property、Flask 的 @app.route 这类「既装饰又参与属性访问」的东西,都从这里出发。

常见坑

坑 1:忘了 functools.wraps。

# ❌ 装饰后 add 的名字和签名都变成 wrapper 的
def log(func):
    def wrapper(*a, **k): return func(*a, **k)
    return wrapper

@log
def add(a, b):
    """两数相加"""
    return a + b

print(add.__name__, add.__doc__)     # 输出: wrapper None


# ✅ 加一行 @functools.wraps(func)(见第三节);另外别忘了 return wrapper,
#    否则装饰后名字直接变成 None,调用报 'NoneType' object is not callable

坑 2:wrapper 不写 *args, **kwargs,self 直接丢。

# ❌ wrapper 不接参数,绑定后的 self 传不进去
def log(func):
    def wrapper(): return func()
    return wrapper

class A:
    @log
    def f(self, x): return x

A().f(1)   # TypeError: wrapper() takes 0 positional arguments but 2 were given
# ✅ 永远写成 (*args, **kwargs) 再原样透传

小结

  • 装饰器 = 接收可调用对象、返回可调用对象的普通函数;闭包负责记状态。
  • @deco 等价于 f = deco(f);不带参两层、带参三层。
  • 写装饰器必加 functools.wraps;类装饰器用 functools.update_wrapper;叠加顺序是「靠近 def 的先包装,外层先进后出」。
  • 缓存的本质是闭包里塞一个 dict。
  • 装饰器返回类实例时要实现 __get__,否则方法调用时 self 会错位。

延伸阅读

  • Python 官方文档 functools(wraps、lru_cache、singledispatch)
  • 《Fluent Python》第 9 章,以及 Flask @app.route 的源码
— 全文完 —回到顶部 ↑
下载推广海报

文章推广海报

《装饰器:从函数到框架(Python 从精通到入门 · 03)》完整推广海报
DISCUSSION

文章回复

0 条公开回复
未登录回复需要审核后公开
还没有回复,欢迎参与讨论。