跳至正文
来两杯美式
返回

Python 知识系列(四):装饰器详解

By 来两杯美式
发布于

1. 为什么要有装饰器

装饰器是一种设计模式,它允许你在不修改原有函数或类的代码的情况下,动态地添加新的功能或修改其行为。 想象一下,你有一个已经完美运行的核心业务逻辑函数,但现在你需要为它添加日志、性能监控或权限检查。如果没有装饰器,你可能需要在原函数内部手动添加这些代码,或者创建一个新的函数来包装旧函数。随着需求的增加,这会导致代码冗余、难以维护。

装饰器通过“包装”被装饰的对象,在对象执行前后或执行时插入额外的逻辑,从而优雅地解决了这个问题。

1.1 Python装饰器与Java注解(Annotation)的异同

相似之处

不同之处

总结: Python装饰器是Python语言中实现“装饰器模式”的一种特殊语法糖,它通过函数包装(或类包装)机制,在运行时修改或增强函数/类的行为。而Java注解更多是提供元数据,行为的实现则依赖于额外的框架和运行时解析机制。

2. 什么是Python装饰器?

2.1 定义和核心原理

Python装饰器是一个函数,它接收另一个函数作为参数,并返回一个新的函数(通常是包装了原函数的新函数)。这个新的函数通常会包含原函数的逻辑,并额外添加一些行为。

用数学公式来表示,如果f是一个函数,decorator是一个装饰器,那么:

@decorator
def f(): ...

等价于:

def f(): ...
f = decorator(f)

也就是说,@符号只是一个语法糖,它将紧跟在它下面的函数定义作为参数传递给@后面的装饰器函数,并将装饰器函数的返回值重新赋值给原函数名。

2.2 基本结构:外部函数、内部函数和被装饰函数

一个典型的函数装饰器通常包含以下三个核心部分:

基本装饰器示例

我们来创建一个简单的装饰器,用于在函数执行前后打印一些信息。

def my_simple_decorator(func):
    """
    这是一个简单的装饰器。
    它接收一个函数 func,并返回一个新函数 wrapper。
    """
    print(f"DEBUG: Decorating function: {func.__name__}") # 装饰器在定义时执行
    def wrapper(*args, **kwargs):
        """
        wrapper 是实际替换原函数被调用的函数。
        它在调用 func 前后添加了额外逻辑。
        """
        print(f"--- Before calling {func.__name__} ---")
        result = func(*args, **kwargs) # 调用原始函数
        print(f"--- After calling {func.__name__} ---")
        return result
    return wrapper

@my_simple_decorator
def say_hello(name):
    """一个简单的打招呼函数"""
    print(f"Hello, {name}!")
    return f"Hello result for {name}"

@my_simple_decorator
def add(a, b):
    """一个简单的加法函数"""
    print(f"Adding {a} and {b}...")
    return a + b

if __name__ == "__main__":
    print("\nCalling say_hello:")
    say_hello("Alice") # 实际调用的是 wrapper 函数
    print("\nCalling add:")
    sum_result = add(10, 20) # 实际调用的是 wrapper 函数
    print(f"Sum result: {sum_result}")
    # 等价于手动包装:
    # def say_hello_original(name):
    #     print(f"Hello, {name}!")
    #     return f"Hello result for {name}"
    # say_hello = my_simple_decorator(say_hello_original)
    # say_hello("Bob")

输出解释

当Python解释器遇到@my_simple_decorator这行时,它会立即调用my_simple_decorator(say_hello)并将返回的wrapper函数绑定到say_hello这个名字上。这就是为什么”DEBUG: Decorating function: say_hello”在程序运行前就打印出来了,因为它发生在函数定义时而非调用时。

3. Closures (闭包) 详解

要想全面理解装饰器,那么闭包这个概念是一定要理解的,它在Python装饰器中扮演着核心角色。

定义: 闭包是一个函数,它“记住了”其被创建时的环境,即使该环境(外部作用域)已经不再活跃,它仍然可以访问和操作该环境中的变量。

在上述装饰器示例中,wrapper函数就是一个闭包。

这是装饰器能够包装和调用原始函数的关键机制。

下面通过一个代码示例,更系统地理解什么是闭包。

# 定义一个方法,返回一个 multiplier 方法
def make_multiplier_of(n):
    """
    外部函数:接收一个乘数 n
    """
    def multiplier(x):
        """
        内部函数:接收一个值 x,并使用外部函数的 n 进行乘法运算
        这是一个闭包,它“记住”了 n 的值
        """
        return x * n
    return multiplier # 外部函数返回内部函数

# 创建一个乘以3的闭包
times3 = make_multiplier_of(3)
# 创建一个乘以5的闭包
times5 = make_multiplier_of(5)

print(f"将 9 乘以 3: {times3(9)}")   # 输出: 27
print(f"将 3 乘以 5: {times5(3)}")   # 输出: 15
print(f"将 2 乘以 3 再乘以 5: {times5(times3(2))}") # 输出: 30

# 可以通过 __closure__ 属性查看闭包捕获的变量
print(f"times3 闭包捕获的变量: {times3.__closure__[0].cell_contents}") # 输出: 3
print(f"times5 闭包捕获的变量: {times5.__closure__[0].cell_contents}") # 输出: 5

4. 为什么使用装饰器?

聊完闭包,再回到装饰器的话题上。装饰器不仅仅是一种语法糖,它更是一种强大的设计模式,为解决软件开发中的常见问题提供了优雅的方案。

4.1 代码复用 (DRY principle)

如果你需要在多个函数中添加相同的辅助功能(例如,所有数据库操作都需要记录日志,所有耗时操作都需要统计时间),你可以将这些公共逻辑封装在一个装饰器中,然后通过@语法轻松地应用到任何需要的函数上,避免了重复编写代码。

4.2 关注点分离 (Separation of Concerns)

核心业务逻辑应该尽可能地纯粹,不应该被横切关注点(Cross-cutting Concerns)所污染。日志、缓存、权限等都是典型的横切关注点。装饰器允许你将这些辅助功能从主函数中剥离出来,保持主函数的简洁和专注于业务逻辑。

# 没有装饰器,日志逻辑分散在各个函数中
def perform_task_without_decorator(data):
    print("LOG: Starting perform_task_without_decorator...")
    # 核心业务逻辑
    result = data * 2
    print("LOG: Finished perform_task_without_decorator.")
    return result

# 有装饰器,日志逻辑集中在装饰器中
@log_calls # 假设有一个日志装饰器
def perform_task_with_decorator(data):
    # 核心业务逻辑
    result = data * 2
    return result

显然,使用装饰器更加清晰和易于维护。

4.3 增强代码可读性与维护性

@decorator语法直观地表明了函数或类被附加了额外的行为,提高了代码的声明性和可读性。当需要修改某个横切关注点的实现时,只需修改装饰器即可,无需改动所有被装饰的函数。

4.4 常见应用场景概览


5. 如何编写和使用基本装饰器

5.1 手动包装函数 (理解 @ 语法糖的本质)

为了更好地理解装饰器,我们首先不使用@语法糖,手动实现函数包装。

def uppercase_decorator(func):
    """
    一个将函数返回值转换为大写的装饰器。
    """
    def wrapper():
        original_result = func()
        return original_result.upper()
    return wrapper

def greet():
    return "hello world"

# 手动应用装饰器
greet = uppercase_decorator(greet)
print(greet()) # 输出: HELLO WORLD

在这个例子中,uppercase_decorator接收greet函数作为参数,返回一个新的wrapper函数。这个wrapper函数在调用原始greet函数后,将结果转换为大写。最后,我们将这个新的wrapper函数重新赋值给greet变量,所以之后所有对greet()的调用实际上都是在调用wrapper()

5.2 @ 语法糖的引入与解释

@语法糖提供了一种更简洁、更Pythonic的方式来应用装饰器。

def uppercase_decorator(func):
    def wrapper():
        original_result = func()
        return original_result.upper()
    return wrapper

@uppercase_decorator # 这行代码等价于 greet = uppercase_decorator(greet)
def greet():
    return "hello world"

print(greet()) # 输出: HELLO WORLD

这里的@uppercase_decorator直接放在greet函数定义的前面,它的作用就相当于greet = uppercase_decorator(greet)。这是Python提供的一种语法便利,使得装饰器的应用更加直观和优雅。


6. 带参数的装饰器

有时,我们希望装饰器本身也能接收参数,从而定制其行为。例如,一个日志装饰器可能需要指定日志级别,或者一个权限装饰器可能需要指定所需角色。

为了实现带参数的装饰器,我们需要引入一个额外的层级:一个“装饰器工厂函数”。这个工厂函数接收装饰器的参数,并返回一个真正的装饰器函数。

工作原理

@decorator_with_args(arg1, arg2)
def f(): ...

等价于:

decorator_instance = decorator_with_args(arg1, arg2)
def f(): ...
f = decorator_instance(f)

所以,decorator_with_args(arg1, arg2)的返回值必须是一个真正的装饰器函数(即一个接收函数作为参数并返回新函数的函数)。

带参数的日志装饰器示例

import logging

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

def log_decorator(log_level):
    """
    这是一个装饰器工厂函数,它接收一个日志级别参数。
    它返回一个实际的装饰器。
    """
    def actual_decorator(func):
        """
        这个是真正的装饰器,它接收被装饰函数 func。
        """
        def wrapper(*args, **kwargs):
            logging.log(log_level, f"Calling function: {func.__name__} with args: {args}, kwargs: {kwargs}")
            try:
                result = func(*args, **kwargs)
                logging.log(log_level, f"Function {func.__name__} returned: {result}")
                return result
            except Exception as e:
                logging.error(f"Function {func.__name__} raised an exception: {e}")
                raise
        return wrapper
    return actual_decorator

@log_decorator(logging.INFO) # 传递参数给装饰器工厂函数
def calculate_product(a, b):
    """计算两个数的乘积"""
    return a * b

@log_decorator(logging.DEBUG) # 另一个日志级别
def fetch_data_from_db(query):
    """模拟从数据库获取数据"""
    logging.debug(f"Executing database query: {query}")
    # 模拟数据获取
    data = {"id": 1, "name": "Python Data", "query": query}
    return data

if __name__ == "__main__":
    print("--- Calling calculate_product ---")
    prod_result = calculate_product(5, 4)
    print(f"Product result: {prod_result}")

    print("\n--- Calling fetch_data_from_db (DEBUG level) ---")
    # 如果logging级别设置为INFO,DEBUG日志不会显示
    # 可以通过 logging.getLogger().setLevel(logging.DEBUG) 来调整
    db_data = fetch_data_from_db("SELECT * FROM users")
    print(f"DB data: {db_data}")

    # 尝试一个会抛出异常的函数
    @log_decorator(logging.ERROR)
    def divide_by_zero(a, b):
        return a / b

    print("\n--- Calling divide_by_zero ---")
    try:
        divide_by_zero(10, 0)
    except ZeroDivisionError:
        print("Caught ZeroDivisionError as expected.")

理解三层嵌套


7. 扩展:functools.wraps 的重要性

当你使用装饰器时,wrapper函数会替换原始函数。这意味着如果你尝试访问被装饰函数的元数据(如函数名__name__、文档字符串__doc__、参数列表等),你将得到wrapper函数的元数据,而不是原始函数的。 这会给调试、文档生成和某些依赖函数元数据的工具带来问题。

为了解决这个问题,Python标准库提供了functools.wraps装饰器。 functools.wraps本身也是一个装饰器,它用于装饰你的wrapper函数,将原始函数的元数据复制到wrapper函数上。

示例对比

from functools import wraps

def no_wraps_decorator(func):
    def wrapper(*args, **kwargs):
        """Wrapper docstring without wraps"""
        return func(*args, **kwargs)
    return wrapper

def with_wraps_decorator(func):
    @wraps(func) # 使用 functools.wraps
    def wrapper(*args, **kwargs):
        """Wrapper docstring with wraps"""
        return func(*args, **kwargs)
    return wrapper

@no_wraps_decorator
def greet_no_wraps(name):
    """Greeter function docstring without wraps"""
    return f"Hello, {name} (no wraps)!"

@with_wraps_decorator
def greet_with_wraps(name):
    """Greeter function docstring with wraps"""
    return f"Hello, {name} (with wraps)!"

if __name__ == "__main__":
    print("--- greet_no_wraps ---")
    print(f"Function name: {greet_no_wraps.__name__}")
    print(f"Function docstring: {greet_no_wraps.__doc__}")
    print(f"Function module: {greet_no_wraps.__module__}")

    print("\n--- greet_with_wraps ---")
    print(f"Function name: {greet_with_wraps.__name__}")
    print(f"Function docstring: {greet_with_wraps.__doc__}")
    print(f"Function module: {greet_with_wraps.__module__}")

输出分析

你会发现greet_no_wraps__name____doc__都显示的是wrapper的信息,而greet_with_wraps则正确显示了原始函数greet_with_wraps的信息。

最佳实践

总是在编写自定义装饰器时,使用@functools.wraps(func)来装饰你的wrapper函数。


8. 类装饰器 (Class Decorators)

除了函数,类也可以作为装饰器,或者被装饰。

8.1 将类作为装饰器:实现 __call__ 方法

如果一个类实现了__call__方法,那么它的实例就可以像函数一样被调用。这种“可调用对象”的特性使得类实例可以作为装饰器使用。

使用类作为装饰器的好处是,你可以利用类的封装性,更好地管理状态(Stateful Decorators)或提供更复杂的逻辑。

状态管理装饰器示例:计数器

from functools import wraps

class CallCounter:
    """
    一个类装饰器,用于统计函数被调用的次数。
    """
    def __init__(self, func):
        """
        构造函数接收被装饰的函数。
        """
        self.func = func
        self.count = 0
        # 同样可以使用 functools.wraps 来复制元数据
        wraps(func)(self) # 将 func 的元数据复制到 CallCounter 实例上

    def __call__(self, *args, **kwargs):
        """
        当被装饰的函数被调用时,实际执行的是这个 __call__ 方法。
        """
        self.count += 1
        print(f"Function '{self.func.__name__}' has been called {self.count} times.")
        return self.func(*args, **kwargs)

@CallCounter
def process_data(data):
    """处理数据"""
    print(f"Processing: {data}")
    return f"Processed: {data}"

@CallCounter
def generate_report():
    """生成报告"""
    print("Generating report...")
    return "Report Generated"

if __name__ == "__main__":
    print("--- Calling process_data ---")
    process_data("initial_data")
    process_data("another_data")
    process_data("final_data")

    print("\n--- Calling generate_report ---")
    generate_report()
    generate_report()

    # 可以直接访问装饰器实例上的状态
    print(f"\nprocess_data was called {process_data.count} times.")
    print(f"generate_report was called {generate_report.count} times.")

工作原理

  1. @CallCounter应用于process_data函数时,Python解释器会执行process_data = CallCounter(process_data)
  2. CallCounter(process_data)会创建一个CallCounter类的实例,并将process_data函数传递给其__init__方法,存储在self.func中。
  3. 这个CallCounter实例现在被赋值给了process_data这个名字。
  4. 当后续代码调用process_data(...)时,实际上是调用了CallCounter实例的__call__方法。

8.2 装饰类

装饰器也可以直接应用于类的定义上,用于修改类的行为,例如添加方法、修改属性、或者将其转换为另一个类(例如,一个单例模式)。

当一个装饰器被应用到类上时,它接收整个类对象作为参数,并期望返回一个新的类或修改后的类。

def add_method_decorator(cls):
    """
    一个类装饰器,为类添加一个新方法。
    """
    def new_say_hello(self):
        return f"Hello from {self.__class__.__name__}!"
    cls.say_hello = new_say_hello
    return cls

@add_method_decorator
class MyClass:
    def __init__(self, name):
        self.name = name
    def greet(self):
        return f"Greetings from {self.name}!"

class AnotherClass:
    pass

if __name__ == "__main__":
    instance1 = MyClass("Instance1")
    print(instance1.greet())
    print(instance1.say_hello()) # 被装饰器添加的方法

    # 未被装饰的类没有这个方法
    instance2 = AnotherClass()
    # instance2.say_hello() # 这会报错:AttributeError: 'AnotherClass' object has no attribute 'say_hello'

另一个常见用例:单例模式

一个类装饰器可以用来将一个普通类转换为单例(Singleton)类。

def singleton(cls):
    """
    单例模式的类装饰器。
    """
    instances = {}
    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"Creating new database connection for {db_name}")
    def connect(self):
        print(f"Connected to {self.db_name}")

if __name__ == "__main__":
    db1 = DatabaseConnection("mydb")
    db2 = DatabaseConnection("mydb")
    db3 = DatabaseConnection("another_db") # 参数不同,依然返回同一个实例
    print(f"db1 is db2: {db1 is db2}") # True
    print(f"db1 is db3: {db1 is db3}") # True (因为实例是第一次创建时通过 cls 创建并存储的,后续调用 get_instance 会直接返回已存储的实例)
    # 【此处不太确定】严格的单例模式通常希望每次调用无论参数如何,都返回同一个实例。
    # 上述实现中,虽然db3使用了不同的参数,但由于get_instance只关心 cls 是否在 instances 中,
    # 导致db3与db1是同一个实例。如果希望不同参数创建不同单例,则需要更复杂的实现,
    # 例如将参数作为key的一部分,或者明确单例不允许带参数。
    # 为了更清晰地展示单例,通常单例类不带或忽略构造函数参数。
    # 如果要确保不同参数创建不同单例,可以这样修改:
    # instances = {}
    # def get_instance(*args, **kwargs):
    #     key = (cls, args, tuple(sorted(kwargs.items()))) # 使用类和参数作为key
    #     if key not in instances:
    #         instances[key] = cls(*args, **kwargs)
    #     return instances[key]
    # return get_instance
    # 这样修改后,db1 is db3 会是 False。但通常单例模式是全局唯一的,不考虑参数差异。

基于单例模式分析类装饰器的调用时机

类装饰器 (Class Decorator) 本质上是一个接收一个类作为参数的函数,并返回一个(通常是修改过的或替换的)类或一个可调用对象。当一个类被装饰时,类装饰器本身只会在类被“定义”或“加载”时执行一次。这里的“加载”通常指的是包含该类的模块被导入(import)时。

下面以上面的单例模式为例,细粒度拆解整体执行流程。

第一阶段:模块加载与 DatabaseConnection 类定义时 (装饰器执行)

当 Python 解释器执行到你的代码时,会发生以下步骤:

  1. 定义 singleton 装饰器函数: 解释器首先加载并定义了 singleton 这个函数。
  2. 定义 DatabaseConnection 类
    • 解释器开始处理 class DatabaseConnection: 块。它会先创建原始的 DatabaseConnection 类对象(包含 __init__, connect 方法)。
    • 遇到 @singleton: Python 发现 DatabaseConnection 类被 @singleton 装饰。
    • 立即执行 singleton 装饰器
      • 解释器会立即调用 singleton(DatabaseConnection)。此时,原始的 DatabaseConnection 类对象 作为 cls 参数传递给 singleton 函数。
      • singleton 函数内部:
        • instances = {} 这个字典被创建。注意: 这是一个局部变量,但由于 get_instance 形成了一个闭包,它会被 get_instance 函数“记住”,并在 get_instance 的多次调用中共享。
        • 嵌套函数 get_instance 被定义。
        • singleton 函数执行完毕,并 返回 get_instance 这个函数对象
      • 替换类名: 此时,关键一步 发生了!全局作用域中的 DatabaseConnection 这个名字,不再指向原始的 DatabaseConnection 类对象,而是被替换成了 singleton 装饰器返回的 get_instance 函数对象。

总结第一阶段: 整个 singleton 装饰器的逻辑(包括 instances 字典的初始化和 get_instance 函数的定义)在 DatabaseConnection 类被定义时就已完成,并且只发生了一次。此时,DatabaseConnection 变量已经是一个函数了。

第二阶段:使用 DatabaseConnection 创建实例时 (装饰器返回的函数执行)

接下来,当你在 if name == "__main__": 块中尝试创建实例时:

  1. db1 = DatabaseConnection("mydb")

    • 你以为你在调用一个类的构造函数,但实际上,由于第一阶段的替换,你现在调用的是 get_instance("mydb")
    • get_instance 函数内部:
      • if cls not in instances: 检查:cls 是原始的 DatabaseConnection 类(闭包记住的)。instances 字典目前是空的,所以条件为真。
      • instances[cls] = cls(*args, **kwargs):这时,原始的 DatabaseConnection.__init__ 方法才首次被调用! DatabaseConnection("mydb") 被执行,并创建了一个真正的 DatabaseConnection 实例。你会在控制台看到 Creating new database connection for mydb
      • 这个新创建的实例被存储在 instances[DatabaseConnection] 中。
      • get_instance 返回这个实例。db1 变量现在指向这个实例。
  2. db2 = DatabaseConnection("mydb")

    • 同样,你调用的是 get_instance("mydb")
    • get_instance 函数内部:
      • if cls not in instances: 检查:cls 仍然是原始的 DatabaseConnection 类。instances 字典中现在已经有 DatabaseConnection 对应的实例了,所以条件为假。
      • return instances[cls]get_instance 直接从 instances 字典中取出之前创建的实例并返回。
      • 注意: 原始的 DatabaseConnection.__init__ 方法 没有再次被调用。你不会再次看到 Creating new database connection for mydb
    • db2 变量现在指向与 db1 完全相同的实例。
  3. db3 = DatabaseConnection("another_db")

    • 同样,你调用的是 get_instance("another_db")
    • get_instance 函数内部:
      • if cls not in instances: 检查:cls 依然是原始的 DatabaseConnection 类。instances 字典中仍然有 DatabaseConnection 对应的实例。条件依然为假。
      • return instances[cls]get_instance 仍然从 instances 字典中取出之前创建的实例并返回。
      • 注意: 即使你传递了不同的参数 another_db,由于这个特定的 singleton 实现是基于 类本身 (cls) 来判断是否已创建实例,而不是基于传递的参数,它仍然返回了同一个实例。如果需要根据参数创建不同的单例,装饰器需要更复杂的逻辑。
    • db3 变量现在指向与 db1db2 完全相同的实例。
# 最终输出
Creating new database connection for mydb
db1 is db2: True
db1 is db3: True

9. 装饰器链 (Chaining Decorators)

一个函数可以被多个装饰器修饰,这被称为装饰器链(Chaining Decorators)或嵌套装饰器。

语法

当一个函数被多个装饰器修饰时,它们会像堆栈一样,自上而下地应用。

@decorator_1
@decorator_2
@decorator_3
def my_function():
    pass

执行顺序

装饰器的应用顺序是从最靠近函数定义的装饰器开始,向外层依次应用。

my_function = decorator_1(decorator_2(decorator_3(my_function)))

这意味着:

  1. decorator_3首先装饰my_function,返回一个新函数f_prime_3
  2. decorator_2接着装饰f_prime_3,返回一个新函数f_prime_2
  3. decorator_1最后装饰f_prime_2,返回最终的函数,它被赋值给my_function

当调用my_function()时,执行流程将从decorator_1wrapper开始,然后是decorator_2wrapper,再是decorator_3wrapper,最后才执行原始的my_function

示例:日志和计时器的组合

import time
import logging
from functools import wraps

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def log_calls(func):
    """日志装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"Entering function: {func.__name__}")
        result = func(*args, **kwargs)
        logging.info(f"Exiting function: {func.__name__}")
        return result
    return wrapper

def timer(func):
    """计时器装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        logging.info(f"Function {func.__name__} took {end_time - start_time:.4f} seconds.")
        return result
    return wrapper

@log_calls # 外层装饰器
@timer     # 内层装饰器,先被应用到 original_complex_operation
def complex_operation(x, y):
    """一个复杂的数学操作"""
    time.sleep(0.05) # 模拟计算时间
    res = x * x + y * y
    logging.info(f"  Performing core logic for complex_operation({x}, {y}) = {res}")
    return res

if __name__ == "__main__":
    print("--- Calling complex_operation ---")
    final_result = complex_operation(3, 4)
    print(f"Final result: {final_result}")

输出分析

从日志输出中,你可以看到log_calls的“Entering”和“Exiting”信息会包围timer的计时信息,这正是因为log_calls是外层装饰器。

最佳实践


10. 实际应用场景与案例

本章将通过一些具体的AI应用场景案例,展示装饰器的实用价值。

10.1 日志装饰器 (Logging)

用于记录函数调用的详细信息,这在调试和监控AI模型训练或推理流程时非常有用。

import logging
from functools import wraps

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def log_function_call(func):
    """记录函数调用、参数、返回值和异常的装饰器。"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        func_name = func.__name__
        args_str = ', '.join(map(repr, args))
        kwargs_str = ', '.join(f"{k}={repr(v)}" for k, v in kwargs.items())
        all_args_str = ', '.join(filter(None, [args_str, kwargs_str]))
        logger.info(f"Calling {func_name}({all_args_str})")
        try:
            result = func(*args, **kwargs)
            logger.info(f"Function {func_name} returned: {repr(result)}")
            return result
        except Exception as e:
            logger.exception(f"Function {func_name} raised an exception: {e}")
            raise
    return wrapper

@log_function_call
def train_model(data_path: str, epochs: int, learning_rate: float = 0.01):
    """模拟AI模型训练过程"""
    logger.info(f"Starting model training with {epochs} epochs...")
    # 模拟训练逻辑
    import random
    accuracy = 0.85 + random.random() * 0.1 # 模拟训练结果
    if random.random() < 0.1: # 模拟10%的几率训练失败
        raise ValueError("Model training failed due to data imbalance.")
    logger.info("Model training completed.")
    return {"accuracy": accuracy, "epochs_run": epochs}

@log_function_call
def predict(model_path: str, input_data: list):
    """模拟AI模型推理过程"""
    logger.info(f"Loading model from {model_path} and predicting for {len(input_data)} items.")
    # 模拟推理逻辑
    predictions = [item * 2 for item in input_data] # 简单模拟
    return predictions

if __name__ == "__main__":
    print("\n--- Model Training ---")
    try:
        train_model(data_path="/data/mnist.csv", epochs=10, learning_rate=0.005)
        train_model(data_path="/data/cifar10.jpg", epochs=5) # 使用默认学习率
    except ValueError as e:
        print(f"Caught expected error: {e}")

    print("\n--- Model Prediction ---")
    predictions = predict(model_path="/models/resnet.pt", input_data=[1, 2, 3, 4])
    print(f"Predictions: {predictions}")

10.2 性能度量装饰器 (Timing)

评估AI算法或数据处理函数的执行效率。

import time
from functools import wraps

def measure_execution_time(func):
    """测量函数执行时间的装饰器。"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter() # 使用perf_counter获取高精度时间
        result = func(*args, **kwargs)
        end_time = time.perf_counter()
        execution_time = end_time - start_time
        print(f"'{func.__name__}' executed in {execution_time:.6f} seconds.")
        return result
    return wrapper

@measure_execution_time
def process_large_dataset(data_size: int):
    """模拟处理大型数据集的函数"""
    print(f"Processing dataset of size {data_size}...")
    # 模拟CPU密集型计算
    _ = [i**2 for i in range(data_size)]
    time.sleep(0.01) # 模拟少量I/O
    print("Dataset processing complete.")
    return f"Processed {data_size} items."

@measure_execution_time
def optimize_hyperparameters(model_config: dict, iterations: int):
    """模拟超参数优化过程"""
    print(f"Optimizing hyperparameters for {iterations} iterations...")
    total_sleep = 0
    for _ in range(iterations):
        sleep_duration = 0.001 * (iterations / 100) # 模拟每次迭代的耗时
        time.sleep(sleep_duration)
        total_sleep += sleep_duration
    print(f"Hyperparameter optimization complete. Total simulated sleep: {total_sleep:.4f}s")
    return {"best_config": {"lr": 0.001, "batch": 32}, "score": 0.92}

if __name__ == "__main__":
    print("\n--- Data Processing ---")
    process_large_dataset(10_000_000)

    print("\n--- Hyperparameter Optimization ---")
    optimize_hyperparameters({"model_type": "CNN"}, 200)

10.3 权限验证装饰器 (Authentication/Authorization)

在AI服务中,某些API可能需要用户具备特定的权限才能访问。

from functools import wraps

# 模拟用户权限数据库
USER_ROLES = {
    "admin": ["read", "write", "delete", "deploy"],
    "analyst": ["read", "report"],
    "guest": ["read"]
}

def require_role(required_roles: list):
    """
    权限验证装饰器工厂函数。
    它接收一个角色列表,并返回一个实际的装饰器。
    """
    def decorator(func):
        @wraps(func)
        def wrapper(user_id: str, *args, **kwargs):
            user_role = USER_ROLES.get(user_id)
            if not user_role:
                raise PermissionError(f"User '{user_id}' not found.")
            # 检查用户是否具备任一所需角色
            if any(role in user_role for role in required_roles):
                print(f"User '{user_id}' with roles {user_role} is authorized for '{func.__name__}'.")
                return func(user_id, *args, **kwargs)
            else:
                raise PermissionError(f"User '{user_id}' does not have required roles ({required_roles}) to access '{func.__name__}'.")
        return wrapper
    return decorator

@require_role(["admin", "deploy"])
def deploy_model(user_id: str, model_name: str, version: str):
    """部署AI模型到生产环境"""
    print(f"User '{user_id}' deploying model '{model_name}' v{version}...")
    return f"Model {model_name} v{version} deployed by {user_id}."

if __name__ == "__main__":
    print("\n--- Model Deployment (Admin) ---")
    try:
        result = deploy_model("admin", "resnet50", "1.0")
        print(result)
    except PermissionError as e:
        print(f"Error: {e}")

    print("\n--- Model Deployment (Analyst - Unauthorized) ---")
    try:
        deploy_model("analyst", "resnet50", "1.0")
    except PermissionError as e:
        print(f"Error: {e}")  # 预期会抛出权限错误

分享这篇文章:
通过邮件分享这篇文章✓ 链接已复制
所属专题
Python
第 4 / 8 篇
查看系列全部文章
  1. 01.Python 知识系列(一):极简核心语法
  2. 02.Python 知识系列(二):面向对象编程
  3. 03.Python 知识系列(三):从 0 开始理解函数
  4. 04.Python 知识系列(四):装饰器详解
  5. 05.Python 知识系列(五):模块和包
  6. 06.Python 知识系列(六):异步编程
  7. 07.Python 知识系列(七):虚拟环境与依赖管理
  8. 08.Python 知识系列(八):企业级开发场景最佳实践

上一篇
Python 知识系列(五):模块和包
下一篇
Python 知识系列(三):从 0 开始理解函数