1. 为什么要有装饰器
装饰器是一种设计模式,它允许你在不修改原有函数或类的代码的情况下,动态地添加新的功能或修改其行为。 想象一下,你有一个已经完美运行的核心业务逻辑函数,但现在你需要为它添加日志、性能监控或权限检查。如果没有装饰器,你可能需要在原函数内部手动添加这些代码,或者创建一个新的函数来包装旧函数。随着需求的增加,这会导致代码冗余、难以维护。
装饰器通过“包装”被装饰的对象,在对象执行前后或执行时插入额外的逻辑,从而优雅地解决了这个问题。
1.1 Python装饰器与Java注解(Annotation)的异同
相似之处:
- 声明式编程: 两者都允许以声明式的方式向代码添加元数据或行为。通过@符号,它们都在代码定义处提供了一种清晰、简洁的方式来表达额外的信息或逻辑。
- 代码解耦: 都能将横切关注点(如日志、事务、权限)与核心业务逻辑分离,提高代码的模块化和可维护性。
- 框架广泛使用: Java的Spring框架大量使用注解实现AOP、依赖注入等;Python的Flask、Django等Web框架也广泛使用装饰器实现路由、权限验证等。
不同之处:
-
本质:
- Java注解: 本质上是元数据。它们本身不包含任何逻辑,只是标记。实际的逻辑需要在运行时通过反射(Reflection)和AOP框架(如AspectJ或Spring AOP)来解析注解并执行相应的处理逻辑。注解是编译时和运行时元数据,其行为由外部处理器定义。
- Python装饰器: 本质上是可调用对象(通常是函数或类)。它直接接收一个函数或类作为输入,然后返回一个新的函数或类。 装饰器本身就包含了要“装饰”行为的逻辑。它是运行时行为的直接修改。
-
灵活性:
- Java注解: 行为是固定的,由注解处理器定义。如果需要改变行为,通常需要修改处理器。
- Python装饰器: 更具动态性。你可以很容易地编写自定义装饰器,甚至创建返回装饰器的函数(带参数的装饰器),以适应不同的需求。
总结: Python装饰器是Python语言中实现“装饰器模式”的一种特殊语法糖,它通过函数包装(或类包装)机制,在运行时修改或增强函数/类的行为。而Java注解更多是提供元数据,行为的实现则依赖于额外的框架和运行时解析机制。
2. 什么是Python装饰器?
2.1 定义和核心原理
Python装饰器是一个函数,它接收另一个函数作为参数,并返回一个新的函数(通常是包装了原函数的新函数)。这个新的函数通常会包含原函数的逻辑,并额外添加一些行为。
用数学公式来表示,如果f是一个函数,decorator是一个装饰器,那么:
@decorator
def f(): ...
等价于:
def f(): ...
f = decorator(f)
也就是说,@符号只是一个语法糖,它将紧跟在它下面的函数定义作为参数传递给@后面的装饰器函数,并将装饰器函数的返回值重新赋值给原函数名。
2.2 基本结构:外部函数、内部函数和被装饰函数
一个典型的函数装饰器通常包含以下三个核心部分:
- 外部函数 (Decorator Function): 这就是“装饰器”本身。它接收被装饰的函数(func)作为参数。
- 内部函数 (Wrapper Function): 在外部函数内部定义,它负责包装原始函数。它会在调用原始函数之前、之后或替代原始函数执行一些额外的逻辑。
- 被装饰函数 (Decorated Function): 即被@decorator语法糖修饰的那个函数。
基本装饰器示例:
我们来创建一个简单的装饰器,用于在函数执行前后打印一些信息。
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函数就是一个闭包。
- 它是在
my_simple_decorator函数内部定义的。 - 它引用了
my_simple_decorator的局部变量func(即被装饰的say_hello或add函数)。 - 当
my_simple_decorator执行完毕并返回wrapper后,即使my_simple_decorator的执行上下文已经结束,wrapper仍然能够访问到它所“封闭”的func变量。
这是装饰器能够包装和调用原始函数的关键机制。
下面通过一个代码示例,更系统地理解什么是闭包。
# 定义一个方法,返回一个 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 常见应用场景概览
- 日志 (Logging):记录函数调用、参数、返回值和异常。
- 性能度量 (Performance Monitoring):测量函数执行时间。
- 权限验证 (Authentication/Authorization):在执行敏感操作前检查用户权限。
- 缓存 (Caching):缓存耗时函数的计算结果,避免重复计算。 Python内置的
functools.lru_cache就是一个常用的缓存装饰器。 - 输入校验 (Input Validation):在函数执行前验证输入参数的合法性。
- 重试机制 (Retry Mechanism):在函数执行失败时自动重试。
- API 注册 (API Registration):将函数注册到某个全局字典或插件系统中。
- 事务管理 (Transaction Management):确保一系列操作的原子性。
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.")
理解三层嵌套:
log_decorator(log_level):最外层函数,接收装饰器参数(如logging.INFO)。它返回actual_decorator。actual_decorator(func):中间层函数,这就是真正的装饰器,它接收被装饰的函数func(如calculate_product)。它返回wrapper。wrapper(*args, **kwargs):最内层函数,它接收被装饰函数的实际调用参数,执行增强逻辑,并调用原始函数。
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.")
工作原理:
- 当
@CallCounter应用于process_data函数时,Python解释器会执行process_data = CallCounter(process_data)。 CallCounter(process_data)会创建一个CallCounter类的实例,并将process_data函数传递给其__init__方法,存储在self.func中。- 这个
CallCounter实例现在被赋值给了process_data这个名字。 - 当后续代码调用
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 解释器执行到你的代码时,会发生以下步骤:
- 定义 singleton 装饰器函数: 解释器首先加载并定义了
singleton这个函数。 - 定义 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__": 块中尝试创建实例时:
-
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变量现在指向这个实例。
- 你以为你在调用一个类的构造函数,但实际上,由于第一阶段的替换,你现在调用的是
-
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完全相同的实例。
- 同样,你调用的是
-
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变量现在指向与db1和db2完全相同的实例。
- 同样,你调用的是
# 最终输出
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)))
这意味着:
decorator_3首先装饰my_function,返回一个新函数f_prime_3。decorator_2接着装饰f_prime_3,返回一个新函数f_prime_2。decorator_1最后装饰f_prime_2,返回最终的函数,它被赋值给my_function。
当调用my_function()时,执行流程将从decorator_1的wrapper开始,然后是decorator_2的wrapper,再是decorator_3的wrapper,最后才执行原始的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}") # 预期会抛出权限错误