Python 装饰器从入门到精通:带参数装饰器与类装饰器的深度实战指南

5次阅读
没有评论

共计 7189 个字符,预计需要花费 18 分钟才能阅读完成。

Python 的装饰器(Decorator)是语言中一个强大而优雅的特性,它允许你在不修改原函数或类代码的情况下,为它们添加额外的功能。无论是日志记录、性能监控、权限校验,还是事务管理,装饰器都能以一种高度可复用和解耦的方式完成这些任务。本文将带你从装饰器基础概念出发,逐步深入到带参数装饰器、类装饰器等高级应用,并通过丰富的实战案例,助你彻底掌握这一 Python 编程利器。

什么是 Python 装饰器?

在深入探讨之前,我们先回顾一下 Python 中一些重要的前置知识:

  • 函数作为一等公民(First-Class Citizens):Python 中的函数可以像普通变量一样被赋值、作为参数传递、作为返回值返回。
  • 闭包(Closures):内部函数可以记住并访问其外部(封闭)作用域的变量,即使外部函数已经执行完毕。
  • 高阶函数(Higher-Order Functions):接收一个或多个函数作为参数,或者返回一个函数的函数。

装饰器本质上就是一个高阶函数,它接收一个函数作为输入,并返回一个经过包装的新函数。这个新函数通常在执行原函数前后添加了一些额外的逻辑。

装饰器基础:手动实现与语法糖

让我们从一个最简单的装饰器开始。假设我们想为某个函数添加一个日志记录的功能。

def my_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"函数 {func.__name__} 即将执行...")
        result = func(*args, **kwargs)
        print(f"函数 {func.__name__} 执行完毕。")
        return result
    return wrapper

def greet(name):
    return f"Hello, {name}!"

# 手动应用装饰器
greet = my_decorator(greet)
print(greet("Alice"))

输出:

 函数 greet 即将执行...
函数 greet 执行完毕。Hello, Alice!

Python 提供了一个更简洁的语法糖 @ 来应用装饰器:

def my_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"函数 {func.__name__} 即将执行...")
        result = func(*args, **kwargs)
        print(f"函数 {func.__name__} 执行完毕。")
        return result
    return wrapper

@my_decorator
def greet(name):
    return f"Hello, {name}!"

print(greet("Bob"))

@my_decorator 等价于 greet = my_decorator(greet)。这种语法使得代码更加清晰和易读。

使用 functools.wraps 保持元数据

一个常见的问题是,当函数被装饰后,它的元数据(如 __name__, __doc__, __module__ 等)会丢失,变成装饰器内部 wrapper 函数的元数据。这在调试和内省时会造成困扰。functools 模块中的 wraps 装饰器可以解决这个问题:

import functools

def my_decorator(func):
    @functools.wraps(func) # 使用 wraps 保持原函数元数据
    def wrapper(*args, **kwargs):
        print(f"函数 {func.__name__} 即将执行...")
        result = func(*args, **kwargs)
        print(f"函数 {func.__name__} 执行完毕。")
        return result
    return wrapper

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

print(calculate_sum(10, 20))
print(f"函数名: {calculate_sum.__name__}")
print(f"文档字符串: {calculate_sum.__doc__}")

此时,calculate_sum.__name__ 将正确显示 calculate_sum,而不是 wrapper。这是一个在编写装饰器时必须遵循的最佳实践。

进阶:带参数装饰器

有时,我们希望装饰器能够接受参数,以实现更灵活的功能。例如,一个日志装饰器可能需要指定日志级别(INFO, DEBUG, ERROR),一个权限验证装饰器可能需要指定所需的角色。这就引出了带参数装饰器。

带参数装饰器的工作原理

带参数装饰器比普通装饰器多一层嵌套。它不再直接返回 wrapper 函数,而是返回一个真正的装饰器。

  1. 最外层函数(装饰器工厂):接收装饰器的参数。
  2. 中间层函数(真正的装饰器):接收被装饰的函数 func
  3. 最内层函数(wrapper):执行装饰逻辑并调用原函数。
import functools

def log_level_decorator(level): # 最外层:接收装饰器参数
    def decorator(func):        # 中间层:接收被装饰的函数
        @functools.wraps(func)
        def wrapper(*args, **kwargs): # 最内层:执行包装逻辑
            if level == "INFO":
                print(f"[INFO] 函数 {func.__name__} 被调用。")
            elif level == "DEBUG":
                print(f"[DEBUG] 函数 {func.__name__} 被调用,参数: {args}, {kwargs}")
            elif level == "ERROR":
                print(f"[ERROR] 函数 {func.__name__} 出现问题。")

            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                print(f"[ERROR] 函数 {func.__name__} 抛出异常: {e}")
                raise
        return wrapper
    return decorator

@log_level_decorator("INFO")
def process_data(data):
    return f"处理数据: {data}"

@log_level_decorator("DEBUG")
def debug_task(a, b):
    return a * b

@log_level_decorator("ERROR")
def might_fail():
    raise ValueError("Something went wrong!")

print(process_data("important_payload"))
print(debug_task(3, 4))
try:
    might_fail()
except ValueError:
    print("捕获到预期错误。")

@log_level_decorator("INFO") 被解析时,它首先调用 log_level_decorator("INFO"),该调用返回一个装饰器函数(即上面代码中的 decorator)。然后,这个返回的装饰器函数再接收 process_data 作为参数,最终返回被包装后的 process_data 函数。

实战案例:带重试机制的装饰器

在网络请求或 I / O 操作中,失败是常态。我们可能希望在失败时自动重试几次。

import functools
import time
import random

def retry(retries=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(retries):
                try:
                    print(f"尝试执行 {func.__name__} (第 {i+1} 次 )...")
                    result = func(*args, **kwargs)
                    print(f"函数 {func.__name__} 执行成功。")
                    return result
                except Exception as e:
                    print(f"函数 {func.__name__} 失败: {e}")
                    if i < retries - 1:
                        time.sleep(delay)
                    else:
                        print(f"函数 {func.__name__} 在 {retries} 次尝试后仍失败。")
                        raise
        return wrapper
    return decorator

@retry(retries=5, delay=2)
def unstable_network_call():
    if random.random() < 0.7: # 70% 的概率失败
        raise ConnectionError("网络连接失败!")
    return "数据已成功获取。"

try:
    print(unstable_network_call())
except ConnectionError as e:
    print(f"最终捕获到错误: {e}")

这个 retry 装饰器接收 retriesdelay 参数,使得我们可以在不同场景下灵活配置重试策略。

深入:类装饰器

装饰器不仅可以是一个函数,也可以是一个类。类装饰器提供了在装饰器中管理状态的强大能力,并且可以用来装饰函数或类本身。

类作为函数装饰器

当一个类被用作函数装饰器时,它需要实现 __init____call__ 方法:

  • __init__:接收被装饰的函数作为参数,并将其保存为实例属性。
  • __call__:使类的实例可以像函数一样被调用,这里面实现了装饰器的逻辑。
import functools

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func) # 同样需要保持元数据
        self.func = func
        self.num_calls = 0

    def __call__(self, *args, **kwargs):
        self.num_calls += 1
        print(f"函数 {self.func.__name__} 已被调用 {self.num_calls} 次。")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello(name):
    print(f"Hello, {name}!")

@CountCalls
def add_numbers(a, b):
    return a + b

say_hello("Alice")
say_hello("Bob")
print(add_numbers(5, 7))
print(add_numbers(1, 2))

print(f"say_hello 函数被调用了 {say_hello.num_calls} 次。")
print(f"add_numbers 函数被调用了 {add_numbers.num_calls} 次。")

这里,CountCalls 的每个实例都维护了一个独立的 num_calls 计数器,这在函数式装饰器中是难以直接实现的(除非使用闭包来保存状态,但类的方式更清晰)。

类作为带参数的函数装饰器

如果类装饰器也需要接收参数,那么 __init__ 方法将首先接收装饰器参数,然后 __call__ 方法将作为真正的装饰器接收被装饰的函数。

import functools

class RateLimiter:
    def __init__(self, calls_per_second): # 装饰器参数
        self.calls_per_second = calls_per_second
        self.last_called = {} # 存储每个函数的上次调用时间

    def __call__(self, func): # 作为装饰器,接收被装饰的函数
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            current_time = time.time()
            if func.__name__ not in self.last_called:
                self.last_called[func.__name__] = 0

            elapsed = current_time - self.last_called[func.__name__]
            if elapsed < (1 / self.calls_per_second):
                wait_time = (1 / self.calls_per_second) - elapsed
                print(f"等待 {wait_time:.2f} 秒,以满足限流要求...")
                time.sleep(wait_time)

            self.last_called[func.__name__] = time.time()
            return func(*args, **kwargs)
        return wrapper

@RateLimiter(calls_per_second=1) # 每秒最多调用一次
def send_email(to_address):
    print(f"向 {to_address} 发送邮件...")
    return "邮件发送成功。"

send_email("[email protected]")
send_email("[email protected]") # 会等待一秒
send_email("[email protected]") # 会等待一秒 

在这个 RateLimiter 例子中,RateLimiter(calls_per_second=1) 首先创建了一个 RateLimiter 类的实例,然后这个实例被用作装饰器,通过其 __call__ 方法来装饰 send_email 函数。

类作为类装饰器

除了装饰函数,装饰器还可以用来装饰一个类。类装饰器通常用于修改或增强类的行为,例如添加方法、属性、验证逻辑,或者实现特定的设计模式(如单例)。

当一个类被用作类装饰器时,它接收一个类作为参数,并返回一个可能已经被修改过的新类。

def singleton(cls):
    """一个将类转换为单例模式的装饰器"""
    instances = {}

    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
        print(f"创建数据库连接: {self.db_name}")

    def connect(self):
        print(f"连接到数据库: {self.db_name}")

# 第一次创建实例
db1 = DatabaseConnection("mydb")
db1.connect()

# 第二次创建实例,将返回同一个实例
db2 = DatabaseConnection("anotherdb") # 注意:这里的 db_name 不会生效,因为返回的是第一次创建的实例
db2.connect()

print(f"db1 is db2: {db1 is db2}")

在这个 singleton 装饰器中,@singletonDatabaseConnection 类作为参数传递给 singleton 函数。singleton 函数返回了一个 get_instance 函数。因此,DatabaseConnection 这个名称现在指向的是 get_instance 函数。每当 DatabaseConnection(...) 被调用时,实际上是调用 get_instance,从而实现了单例模式。

高级主题与最佳实践

装饰器链

可以为一个函数应用多个装饰器,它们会从内到外依次执行:

@decorator_outer
@decorator_inner
def my_function():
    pass

等价于 my_function = decorator_outer(decorator_inner(my_function))

类型提示

为带参数装饰器编写准确的类型提示,可以提高代码的可读性和可维护性,特别是在大型项目中。这通常需要使用 typing.Callabletyping.ParamSpec(Python 3.10+)或 typing.TypeVar

装饰器在框架中的应用

Python 装饰器在许多流行框架中都有广泛应用:

  • Flask/Django@app.route() 用于 URL 路由,@login_required 用于权限验证。
  • SQLAlchemy@event.listens_for() 用于事件监听。
  • pytest@pytest.mark.parametrize() 用于参数化测试。

理解装饰器原理,能让你更好地理解和使用这些框架。

注意事项

  • 避免过度复杂 :装饰器虽然强大,但过度使用或设计过于复杂的装饰器可能降低代码的可读性。
  • 执行顺序 :理解装饰器链的执行顺序至关重要。
  • 异常处理 :在 wrapper 函数中妥善处理异常,确保原函数的异常能够被正确传递或处理。
  • functools.partial:有时,可以使用 functools.partial 来部分应用函数,这在某些情况下可以作为装饰器的替代方案或辅助工具。

总结

Python 装饰器是实现 AOP(面向切面编程)思想的绝佳工具,它能够帮助我们以声明式的方式增强函数和类的功能,提高代码的模块化、可读性和可维护性。从基本的函数装饰器到带参数装饰器,再到功能强大的类装饰器,每一种形式都提供了独特的优势。

通过本文的讲解与实战案例,相信你已经对 Python 装饰器有了从入门到精通的理解。现在,是时候将这些知识应用到你的项目中,编写出更优雅、更高效的 Python 代码了!

正文完
 0
评论(没有评论)