Python装饰器原理与实战

装饰器（Decorator）是Python中最优雅和强大的特性之一。它本质上是一个函数或类，用于在不修改原函数代码的前提下，为函数添加额外功能。本文将从函数作为一等公民和闭包概念讲起，逐步深入到各类装饰器的原理与实战应用。

一、函数是一等公民

在Python中，函数是一等公民（First-Class Citizen），意味着函数可以像普通对象一样被赋值给变量、作为参数传递、作为返回值返回，以及存储在数据结构中。

# 函数可以赋值给变量
def greet(name):
    return f"你好，{name}！"

say_hello = greet  # 将函数赋值给变量
print(say_hello("平平"))  # 你好，平平！

# 函数可以作为参数传递
def call_function(func, *args):
    """调用传入的函数"""
    return func(*args)

result = call_function(greet, "世界")
print(result)  # 你好，世界！

# 函数可以作为返回值
def get_multiplier(factor):
    """返回一个乘法函数"""
    def multiplier(x):
        return x * factor
    return multiplier

double = get_multiplier(2)
triple = get_multiplier(3)

print(double(5))   # 10
print(triple(5))   # 15

# 函数可以存储在列表中
operations = [lambda x: x + 1, lambda x: x * 2, lambda x: x ** 2]
for op in operations:
    print(op(3))  # 4, 6, 9

二、闭包概念

闭包（Closure）是一个函数对象，它记住了定义时所在作用域中的变量，即使这个作用域已经结束。闭包是理解装饰器的关键基础。

def make_counter():
    """创建一个计数器闭包"""
    count = 0  # 自由变量

    def counter():
        nonlocal count  # 声明使用外层函数的变量
        count += 1
        return count

    return counter

# 每个调用创建独立的闭包
counter_a = make_counter()
counter_b = make_counter()

print(counter_a())  # 1
print(counter_a())  # 2
print(counter_a())  # 3
print(counter_b())  # 1（独立计数）
print(counter_b())  # 2

# 闭包的实际应用：带缓存的斐波那契
def fibonacci():
    cache = {0: 0, 1: 1}

    def fib(n):
        if n not in cache:
            cache[n] = fib(n - 1) + fib(n - 2)
        return cache[n]

    return fib

fib = fibonacci()
print(fib(10))   # 55
print(fib(50))   # 12586269025
print(fib(100))  # 354224848179261915075

三、装饰器基础语法

装饰器的核心思想是：接受一个函数作为参数，返回一个增强版的新函数。使用 @语法糖可以简洁地将装饰器应用到目标函数上。

# 装饰器的基本结构
def my_decorator(func):
    def wrapper(*args, **kwargs):
        # 在函数执行前添加功能
        print(f"[调用前] 正在执行 {func.__name__}")
        # 执行原函数
        result = func(*args, **kwargs)
        # 在函数执行后添加功能
        print(f"[调用后] {func.__name__} 执行完毕")
        return result
    return wrapper

# 使用装饰器（等价于 say_hi = my_decorator(say_hi)）
@my_decorator
def say_hi(name):
    print(f"Hi, {name}!")
    return f"已问候 {name}"

print(say_hi("平平"))
# 输出：
# [调用前] 正在执行 say_hi
# Hi, 平平!
# [调用后] say_hi 执行完毕
# 已问候 平平

四、使用functools.wraps保留元信息

上面的装饰器有一个问题：被装饰后的函数丢失了原始函数的元信息（如 __name__、__doc__ 等）。functools.wraps 可以解决这个问题。

from functools import wraps

def log_decorator(func):
    """带日志功能的装饰器"""
    @wraps(func)  # 保留原函数的元信息
    def wrapper(*args, **kwargs):
        print(f"[LOG] 调用函数：{func.__name__}")
        print(f"[LOG] 参数：args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        print(f"[LOG] 返回值：{result}")
        return result
    return wrapper

@log_decorator
def add(a, b):
    """计算两个数的和"""
    return a + b

@log_decorator
def multiply(a, b):
    """计算两个数的乘积"""
    return a * b

print(add.__name__)   # add（保留了原函数名）
print(add.__doc__)    # 计算两个数的和（保留了文档字符串）
print("---")
result = add(3, 5)
print(f"结果：{result}")
print("---")
result2 = multiply(4, 6)
print(f"结果：{result2}")

五、带参数的装饰器

有时装饰器自身也需要参数。这需要三层嵌套：外层函数接收装饰器参数，中间层接收被装饰函数，最内层是实际的包装函数。

from functools import wraps
import time

def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
    """重试装饰器：在遇到异常时自动重试"""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        print(f"[重试] {func.__name__} 失败 {max_attempts} 次，放弃")
                        raise
                    print(f"[重试] 第{attempt}次失败：{e}，{delay}秒后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

# 使用带参数的装饰器
@retry(max_attempts=3, delay=0.5, exceptions=(ValueError, TypeError))
def parse_int(text):
    """将文本解析为整数"""
    return int(text)

# 模拟不稳定的网络请求
import random

@retry(max_attempts=5, delay=1)
def fetch_data(url):
    """模拟网络请求（随机失败）"""
    if random.random() < 0.7:
        raise ConnectionError("网络连接失败")
    return {"status": "ok", "data": "Hello from server"}

print(parse_int("42"))      # 42
print(parse_int("abc"))     # 会重试3次后报错
# print(fetch_data("http://example.com"))  # 可能会重试多次

六、类装饰器

除了函数装饰器，还可以使用类来实现装饰器。类装饰器通常需要实现 __init__ 和 __call__ 方法。

import time
from functools import wraps

class CountCalls:
    """统计函数调用次数的类装饰器"""

    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"[统计] {self.func.__name__} 第 {self.count} 次调用")
        return self.func(*args, **kwargs)

    def reset(self):
        """重置计数"""
        self.count = 0

@CountCalls
def calculate_pi(terms):
    """用莱布尼茨公式计算圆周率"""
    pi = 0
    for i in range(terms):
        pi += (-1) ** i / (2 * i + 1)
    return pi * 4

print(f"Pi(100) = {calculate_pi(100):.6f}")
print(f"Pi(1000) = {calculate_pi(1000):.10f}")
print(f"Pi(10000) = {calculate_pi(10000):.12f}")


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

    def __init__(self, precision=4):
        self.precision = precision

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            elapsed = time.perf_counter() - start
            print(f"[计时] {func.__name__} 耗时 {elapsed:.{self.precision}f}秒")
            return result
        return wrapper

@Timer(precision=6)
def slow_sum(n):
    """慢速求和（故意使用循环）"""
    total = 0
    for i in range(n):
        total += i
    return total

print(f"结果：{slow_sum(1000000)}")

七、常用内置装饰器

Python自带了几个非常实用的装饰器：@property、@staticmethod、@classmethod、@functools.lru_cache 等。

# ========== @property：将方法变为属性访问 ==========
class Rectangle:
    def __init__(self, width, height):
        self._width = width
        self._height = height

    @property
    def width(self):
        return self._width

    @width.setter
    def width(self, value):
        if value <= 0:
            raise ValueError("宽度必须大于0")
        self._width = value

    @property
    def height(self):
        return self._height

    @height.setter
    def height(self, value):
        if value <= 0:
            raise ValueError("高度必须大于0")
        self._height = value

    @property
    def area(self):
        """只读属性：面积"""
        return self._width * self._height

    @property
    def perimeter(self):
        """只读属性：周长"""
        return 2 * (self._width + self._height)

rect = Rectangle(10, 5)
print(f"面积：{rect.area}")
print(f"周长：{rect.perimeter}")
rect.width = 20
print(f"新面积：{rect.area}")

# ========== @functools.lru_cache：缓存装饰器 ==========
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_computation(x):
    """模拟耗时计算"""
    print(f"  正在计算 {x}...")
    import time
    time.sleep(0.1)
    return x ** 2 + x + 1

print("第一次调用 expensive_computation(5):")
print(expensive_computation(5))   # 会实际计算

print("第二次调用 expensive_computation(5):")
print(expensive_computation(5))   # 从缓存返回，不计算

print("缓存信息：", expensive_computation.cache_info())

# ========== @functools.singledispatch：单分派泛型函数 ==========
from functools import singledispatch

@singledispatch
def process_data(data):
    """默认处理"""
    print(f"默认处理：{data}")

@process_data.register
def _(data: int):
    print(f"整数处理：{data} 的平方是 {data ** 2}")

@process_data.register
def _(data: str):
    print(f"字符串处理：'{data}' 长度为 {len(data)}")

@process_data.register(list)
def _(data):
    print(f"列表处理：包含 {len(data)} 个元素")

process_data(42)
process_data("hello")
process_data([1, 2, 3])
process_data(3.14)  # 走默认

八、实战案例：计时器装饰器

import time
import functools

def timer(func):
    """精确计时装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            result = func(*args, **kwargs)
            return result
        finally:
            elapsed = time.perf_counter() - start
            if elapsed < 1:
                unit = "毫秒"
                elapsed *= 1000
            else:
                unit = "秒"
            print(f"[{func.__name__}] 执行耗时: {elapsed:.2f}{unit}")
    return wrapper

@timer
def bubble_sort(arr):
    """冒泡排序"""
    n = len(arr)
    for i in range(n):
        for j in range(0, n - i - 1):
            if arr[j] > arr[j + 1]:
                arr[j], arr[j + 1] = arr[j + 1], arr[j]
    return arr

import random
data = random.sample(range(10000), 1000)
bubble_sort(data.copy())

九、实战案例：日志装饰器

import functools
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def log_execution(level=logging.INFO):
    """日志装饰器，可指定日志级别"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            logger = logging.getLogger(func.__module__)
            func_name = func.__name__

            logger.log(level, f">>> 开始执行 {func_name}")
            logger.log(level, f"    参数: args={args}, kwargs={kwargs}")

            try:
                result = func(*args, **kwargs)
                logger.log(level, f"<<< {func_name} 执行成功")
                logger.log(level, f"    返回值类型: {type(result).__name__}")
                return result
            except Exception as e:
                logger.error(f"!!! {func_name} 执行异常: {e}")
                raise
        return wrapper
    return decorator

@log_execution(level=logging.INFO)
def divide(a, b):
    """除法运算"""
    return a / b

print(divide(10, 3))
print("---")
try:
    divide(10, 0)
except ZeroDivisionError:
    print("捕获到除零错误")

十、实战案例：缓存装饰器

import functools
import hashlib
import pickle
import time

def timed_cache(max_age_seconds=60):
    """带过期时间的缓存装饰器"""

    def decorator(func):
        cache = {}

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key_parts = [func.__name__, str(args), str(sorted(kwargs.items()))]
            key = hashlib.md5("|".join(key_parts).encode()).hexdigest()

            # 检查缓存
            if key in cache:
                result, timestamp = cache[key]
                if time.time() - timestamp < max_age_seconds:
                    print(f"[缓存命中] {func.__name__} 缓存有效")
                    return result
                else:
                    print(f"[缓存过期] {func.__name__} 重新计算")
                    del cache[key]

            # 执行函数并缓存
            result = func(*args, **kwargs)
            cache[key] = (result, time.time())
            print(f"[缓存写入] {func.__name__} 结果已缓存")
            return result

        def clear_cache():
            """清除缓存"""
            cache.clear()
            print(f"[缓存清除] {func.__name__} 缓存已清空")

        wrapper.clear_cache = clear_cache
        return wrapper

    return decorator

@timed_cache(max_age_seconds=10)
def get_weather(city):
    """模拟获取天气数据（带缓存）"""
    print(f"  [模拟请求] 正在获取 {city} 的天气...")
    time.sleep(0.5)  # 模拟网络延迟
    weather_data = {
        "北京": {"temp": 28, "weather": "晴"},
        "上海": {"temp": 32, "weather": "多云"},
        "广州": {"temp": 35, "weather": "雷阵雨"},
    }
    return weather_data.get(city, {"temp": 25, "weather": "未知"})

print(get_weather("北京"))
print(get_weather("北京"))  # 缓存命中
time.sleep(11)
print(get_weather("北京"))  # 缓存过期，重新请求
get_weather.clear_cache()

十一、装饰器叠加与执行顺序

多个装饰器可以叠加使用，执行顺序是从下往上（靠近函数的先执行）。

import functools

def bold(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

def italic(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

def underline(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

# 装饰器从下往上执行
# 等价于: text = bold(italic(underline(text)))
@bold
@italic
@underline
def text():
    return "Hello, Decorator!"

print(text())
# 输出: Hello, Decorator!

装饰器是Python中非常强大的工具，理解其原理后，你可以在日志记录、性能监控、权限验证、缓存、重试机制等众多场景中灵活运用。建议在实际项目中多加练习，逐步掌握各种装饰器的使用技巧。