跳至主要內容

Python 装饰器深度实战——参数化装饰器、类装饰器、functools 工具链

老张2026/4/30大约 7 分钟

Python 装饰器深度实战——参数化装饰器、类装饰器、functools 工具链

适读人群：会基础装饰器、想深入掌握装饰器高级用法的 Python 工程师 | 阅读时长：约 16 分钟 | 核心价值：彻底搞懂装饰器的各种形态，写出可复用的 AOP 式代码

装饰器让我少写了一半代码

接触 Python 装饰器之前，我写过一段很"Java"的代码——每个 API 函数都要手动写日志、计时、重试逻辑：

def call_llm_api(prompt: str) -> str:
    start = time.time()
    logger.info(f"开始调用 LLM: {prompt[:50]}")
    for attempt in range(3):
        try:
            result = _do_call(prompt)
            logger.info(f"调用成功，耗时 {time.time()-start:.2f}s")
            return result
        except Exception as e:
            if attempt == 2:
                raise
            logger.warning(f"第{attempt+1}次失败，重试: {e}")
            time.sleep(1)

这段逻辑在 10 个函数里重复了 10 遍。后来我用装饰器重构：

@log_call
@retry(times=3, delay=1.0)
@timer
def call_llm_api(prompt: str) -> str:
    return _do_call(prompt)

三行声明，所有横切关注点分离出去，业务代码干净了。这就是装饰器的威力。

一、装饰器基础：它究竟是什么

# 装饰器本质上是一个函数，接收函数，返回函数
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("函数执行前")
        result = func(*args, **kwargs)
        print("函数执行后")
        return result
    return wrapper

# @语法糖等价于：
# say_hello = my_decorator(say_hello)
@my_decorator
def say_hello(name: str) -> str:
    return f"你好，{name}！"

踩坑实录 1：不加 @functools.wraps 导致函数元数据丢失

import functools

# 错误：不加 wraps，函数名和文档字符串被覆盖
def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def my_func():
    """这是我的函数"""
    pass

print(my_func.__name__)   # 'wrapper'（不是 'my_func'！）
print(my_func.__doc__)    # None（文档字符串丢失！）

# 正确：必须加 @functools.wraps(func)
def good_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@good_decorator
def my_func():
    """这是我的函数"""
    pass

print(my_func.__name__)   # 'my_func'（正确）
print(my_func.__doc__)    # '这是我的函数'（正确）

二、参数化装饰器：工厂模式

参数化装饰器需要三层嵌套——最外层接收参数，中间层接收函数，最内层是实际包装：

import functools
import time
from typing import Callable, TypeVar

F = TypeVar("F", bound=Callable)


def retry(
    times: int = 3,
    delay: float = 1.0,
    exceptions: tuple[type[Exception], ...] = (Exception,),
):
    """重试装饰器"""
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < times:
                        print(f"第 {attempt} 次失败，{delay}s 后重试: {e}")
                        time.sleep(delay)
                    else:
                        print(f"全部 {times} 次均失败")
            raise last_exc
        return wrapper  # type: ignore
    return decorator


def rate_limit(calls_per_second: float = 10.0):
    """简单限流装饰器"""
    min_interval = 1.0 / calls_per_second
    last_call_time = [0.0]  # 用列表包裹，让闭包可以修改

    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            elapsed = now - last_call_time[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call_time[0] = time.monotonic()
            return func(*args, **kwargs)
        return wrapper  # type: ignore
    return decorator


def cache_result(ttl_seconds: int = 60):
    """带 TTL 的结果缓存装饰器"""
    def decorator(func: F) -> F:
        _cache: dict = {}

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            if key in _cache:
                result, expire_at = _cache[key]
                if time.monotonic() < expire_at:
                    return result
            result = func(*args, **kwargs)
            _cache[key] = (result, time.monotonic() + ttl_seconds)
            return result
        return wrapper  # type: ignore
    return decorator

三、类装饰器：用类实现装饰器

类装饰器通过实现 __call__ 方法来当做装饰器使用，适合有状态的装饰器：

import functools
import threading
import time


class Timer:
    """计时装饰器（类实现）"""

    def __init__(self, func: Callable):
        functools.update_wrapper(self, func)
        self.func = func
        self.call_count = 0
        self.total_time = 0.0

    def __call__(self, *args, **kwargs):
        self.call_count += 1
        start = time.perf_counter()
        try:
            result = self.func(*args, **kwargs)
            return result
        finally:
            elapsed = time.perf_counter() - start
            self.total_time += elapsed

    @property
    def avg_time(self) -> float:
        return self.total_time / self.call_count if self.call_count else 0.0

    def stats(self) -> dict:
        return {
            "function": self.func.__name__,
            "call_count": self.call_count,
            "total_time": round(self.total_time, 4),
            "avg_time": round(self.avg_time, 4),
        }


class SingletonDecorator:
    """单例装饰器：确保类只有一个实例"""

    def __init__(self, cls):
        functools.update_wrapper(self, cls)
        self._cls = cls
        self._instance = None
        self._lock = threading.Lock()

    def __call__(self, *args, **kwargs):
        if self._instance is None:
            with self._lock:
                if self._instance is None:
                    self._instance = self._cls(*args, **kwargs)
        return self._instance


@SingletonDecorator
class DatabasePool:
    def __init__(self, size: int = 10):
        self.size = size
        print(f"创建连接池，大小: {size}")

四、functools 工具链

4.1 functools.lru_cache

import functools

@functools.lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# 查看缓存信息
print(fibonacci.cache_info())  # CacheInfo(hits=..., misses=..., maxsize=128, currsize=...)
fibonacci.cache_clear()  # 清除缓存

# Python 3.9+：无大小限制的缓存
@functools.cache
def expensive_compute(n: int) -> int:
    return sum(range(n))

4.2 functools.partial：偏函数

from functools import partial

def power(base: float, exponent: float) -> float:
    return base ** exponent

# 创建特化版本
square = partial(power, exponent=2)
cube = partial(power, exponent=3)

print(square(4))  # 16.0
print(cube(3))    # 27.0

# 实际应用：固定部分参数的 API 调用
import httpx

def api_request(url: str, method: str = "GET", timeout: int = 30, **kwargs):
    with httpx.Client(timeout=timeout) as client:
        return client.request(method, url, **kwargs)

# 特化为内部 API 调用（固定 base_url 和超时）
internal_api = partial(api_request, timeout=5)

4.3 functools.reduce

from functools import reduce
from operator import add, mul

numbers = [1, 2, 3, 4, 5]
total = reduce(add, numbers)           # 15
product = reduce(mul, numbers)         # 120
max_val = reduce(lambda a, b: a if a > b else b, numbers)  # 5

五、完整可运行示例：装饰器工具库

#!/usr/bin/env python3
"""
Python 装饰器工具库完整示例
包含：参数化重试、计时、缓存、日志、限流装饰器
"""
import functools
import logging
import time
from typing import Any, Callable, TypeVar

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

F = TypeVar("F", bound=Callable[..., Any])


# ===== 1. 重试装饰器 =====
def retry(times: int = 3, delay: float = 0.5, exceptions=(Exception,)):
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == times:
                        raise
                    logger.warning(f"[retry] {func.__name__} 第{attempt}次失败: {e}，{delay}s 后重试")
                    time.sleep(delay)
        return wrapper  # type: ignore
    return decorator


# ===== 2. 计时装饰器 =====
def timer(func: F) -> F:
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = (time.perf_counter() - start) * 1000
        logger.info(f"[timer] {func.__name__} 耗时 {elapsed:.2f}ms")
        return result
    return wrapper  # type: ignore


# ===== 3. 带 TTL 缓存装饰器 =====
def ttl_cache(seconds: int = 60):
    def decorator(func: F) -> F:
        _store: dict = {}

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            if key in _store:
                value, exp = _store[key]
                if time.monotonic() < exp:
                    logger.debug(f"[cache] {func.__name__} 命中缓存")
                    return value
            value = func(*args, **kwargs)
            _store[key] = (value, time.monotonic() + seconds)
            logger.debug(f"[cache] {func.__name__} 写入缓存")
            return value
        return wrapper  # type: ignore
    return decorator


# ===== 4. 日志装饰器 =====
def log_call(func: F) -> F:
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        arg_str = ", ".join([repr(a) for a in args[:3]])
        logger.info(f"[call] {func.__name__}({arg_str}...)")
        try:
            result = func(*args, **kwargs)
            logger.info(f"[call] {func.__name__} 返回: {repr(result)[:50]}")
            return result
        except Exception as e:
            logger.error(f"[call] {func.__name__} 异常: {e}")
            raise
    return wrapper  # type: ignore


# ===== 5. 组合使用 =====
call_count = [0]

@log_call
@timer
@retry(times=3, delay=0.1, exceptions=(ValueError,))
@ttl_cache(seconds=10)
def fetch_price(product_id: int) -> float:
    """模拟获取商品价格，前两次失败"""
    call_count[0] += 1
    if call_count[0] <= 2:
        raise ValueError(f"临时服务不可用 (调用 #{call_count[0]})")
    return 299.0 + product_id * 10


def main():
    print("=== 装饰器工具库演示 ===\n")

    # 演示重试：前两次失败，第三次成功
    print("--- 第1次调用（会重试）---")
    try:
        price = fetch_price(1)
        print(f"价格: {price}")
    except Exception as e:
        print(f"最终失败: {e}")

    # 重置计数
    call_count[0] = 0

    print("\n--- 重置后，第1次调用 ---")
    price = fetch_price(1)
    print(f"价格: {price}")

    print("\n--- 第2次调用（命中缓存）---")
    price = fetch_price(1)
    print(f"价格（来自缓存）: {price}")

    # 演示 partial
    from functools import partial
    print("\n--- functools.partial 演示 ---")
    get_discounted = partial(lambda price, pct: price * (1 - pct), pct=0.2)
    original_prices = [100, 200, 350]
    discounted = list(map(get_discounted, original_prices))
    print(f"原价: {original_prices}")
    print(f"8折价: {discounted}")

    # lru_cache 演示
    @functools.lru_cache(maxsize=32)
    def fib(n: int) -> int:
        return n if n < 2 else fib(n-1) + fib(n-2)

    print("\n--- lru_cache Fibonacci ---")
    print(f"fib(30) = {fib(30)}")
    print(f"缓存信息: {fib.cache_info()}")


if __name__ == "__main__":
    main()

六、踩坑实录 2：装饰器顺序影响执行顺序

# 装饰器从下到上应用（从内到外包裹）
@decorator_a  # 最后应用，最外层
@decorator_b  # 先应用
def my_func():
    pass

# 等价于：my_func = decorator_a(decorator_b(my_func))
# 调用顺序：decorator_a 的前置代码 → decorator_b 的前置代码 → my_func → decorator_b 的后置代码 → decorator_a 的后置代码

# 实际例子：log_call 和 timer 的顺序
@log_call  # 外层：先打印调用日志，最后打印返回值
@timer     # 内层：只计时 my_func 本身
def my_func():
    pass
# 日志包含了计时信息（因为 log_call 在外层）

七、踩坑实录 3：装饰 async 函数时忘记 await

# 错误：装饰器里的 wrapper 是同步的，不能装饰 async 函数
def bad_timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)  # async 函数返回协程，不是结果！
        elapsed = time.time() - start
        print(f"耗时: {elapsed:.2f}s")
        return result
    return wrapper

# 正确：检测是否是 async 函数，使用 async wrapper
import asyncio

def smart_timer(func):
    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = await func(*args, **kwargs)
        print(f"[timer] {func.__name__} 耗时 {(time.perf_counter()-start)*1000:.2f}ms")
        return result

    @functools.wraps(func)
    def sync_wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        print(f"[timer] {func.__name__} 耗时 {(time.perf_counter()-start)*1000:.2f}ms")
        return result

    return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

总结

装饰器的四种形态：

  1. 简单装饰器：def decorator(func): ... return wrapper
  2. 参数化装饰器：三层嵌套，def decorator_factory(...): def decorator(func): ...
  3. 类装饰器：实现 __call__，适合有状态的场景
  4. functools 工具链：lru_cache/cache、partial、reduce、wraps

三个铁律：永远加 @functools.wraps(func)，异步函数要用 async wrapper，装饰器顺序从下到上。

编辑此页
最近更新:
贡献者: xiaozhangge