Python装饰器（Decorator）是Python语言中一个强大而优雅的特性，它允许我们在不修改原有函数代码的情况下，为函数添加额外的功能。装饰器本质上是一个高阶函数，它接收一个函数作为参数并返回一个新的函数。在本文中，我们将从基础概念开始，逐步深入探讨装饰器的工作原理、常见用法以及高级应用场景。
什么是装饰器及其基本原理
装饰器是Python中实现面向切面编程（AOP）的核心工具，它遵循开放封闭原则，即对扩展开放，对修改封闭。装饰器通过使用Python的函数作为一等公民特性，以及闭包的概念，实现了对函数行为的动态扩展。
装饰器的基本语法
在Python中，装饰器使用@符号表示，它实际上是语法糖。让我们从一个简单的例子开始：
def my_decorator(func):
def wrapper():
print("在函数调用前执行")
func()
print("在函数调用后执行")
return wrapper
@my_decorator
def say_hello():
print("Hello!")
# 调用装饰后的函数
say_hello()
输出结果：
在函数调用前执行
Hello!
在函数调用后执行
这个例子展示了装饰器的基本工作原理。@my_decorator语法等价于执行say_hello = my_decorator(say_hello)。装饰器函数接收被装饰的函数作为参数，返回一个新的函数（包装函数），这个新函数包含了额外的功能和对原函数的调用。
装饰器的工作机制
让我们通过更详细的代码来理解装饰器的执行过程：
def debug_decorator(func):
print(f"装饰器被调用，装饰函数: {func.__name__}")
def wrapper(*args, **kwargs):
print(f"开始执行 {func.__name__}，参数: {args}, {kwargs}")
result = func(*args, **kwargs)
print(f"函数 {func.__name__} 执行完成，返回值: {result}")
return result
print(f"返回包装函数 wrapper")
return wrapper
@debug_decorator
def add(a, b):
return a + b
# 此时装饰器已经执行，add现在指向wrapper
print("准备调用add函数")
result = add(3, 5)
print(f"最终结果: {result}")
输出：
装饰器被调用，装饰函数: add
返回包装函数 wrapper
准备调用add函数
开始执行 add，参数: (3, 5), {}
函数 add 执行完成，返回值: 8
最终结果: 8
这个例子清楚地展示了装饰器的执行时机：在Python解释器加载模块时，装饰器函数就会执行，返回的包装函数会替换原函数名。每次调用被装饰的函数时，实际上是在调用包装函数。
带参数的装饰器
在实际应用中，我们经常需要让装饰器本身接受参数，这就需要创建一个返回装饰器的函数，形成三层嵌套结构。
装饰器参数的实现
def repeat(times):
"""重复执行函数的装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
for i in range(times):
print(f"第 {i+1} 次执行:")
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@repeat(3)
def greet(name):
print(f"你好, {name}!")
return f"问候了 {name}"
# 调用函数
result = greet("张三")
print(f"返回值: {result}")
输出：
第 1 次执行:
你好, 张三!
第 2 次执行:
你好, 张三!
第 3 次执行:
你好, 张三!
返回值: 问候了 张三
带默认参数的装饰器
def validate_input(required_type=None, min_length=None, max_length=None):
"""参数验证装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
# 验证位置参数
for i, arg in enumerate(args):
if required_type and not isinstance(arg, required_type):
raise TypeError(f"参数 {i} 应该是 {required_type} 类型，但得到 {type(arg)}")
if min_length and hasattr(arg, '__len__') and len(arg) < min_length:
raise ValueError(f"参数 {i} 的长度应该至少为 {min_length}")
if max_length and hasattr(arg, '__len__') and len(arg) > max_length:
raise ValueError(f"参数 {i} 的长度应该不超过 {max_length}")
# 验证关键字参数
for key, value in kwargs.items():
if required_type and not isinstance(value, required_type):
raise TypeError(f"参数 {key} 应该是 {required_type} 类型，但得到 {type(value)}")
return func(*args, **kwargs)
return wrapper
return decorator
@validate_input(required_type=str, min_length=1, max_length=50)
def create_user(username, email):
print(f"创建用户: {username}, 邮箱: {email}")
return {"username": username, "email": email}
# 正常调用
user = create_user("Alice", "alice@example.com")
print(user)
# 异常调用示例
try:
create_user("Bob", "") # 空字符串会触发min_length验证
except ValueError as e:
print(f"验证错误: {e}")
try:
create_user("Charlie", 123) # 数字会触发类型验证
except TypeError as e:
print(f"类型错误: {e}")
类装饰器
除了函数装饰器，Python还支持类装饰器。类装饰器需要实现__call__方法，使其实例可调用。
基础类装饰器
class CountCalls:
"""统计函数调用次数的类装饰器"""
def __init__(self, func):
self.func = func
self.call_count = 0
def __call__(self, *args, **kwargs):
self.call_count += 1
print(f"函数 {self.func.__name__} 被调用了 {self.call_count} 次")
return self.func(*args, **kwargs)
def __get__(self, instance, owner):
"""支持实例方法的绑定"""
import functools
return functools.partial(self.__call__, instance)
@CountCalls
def process_data(data):
print(f"处理数据: {data}")
return len(data)
# 测试调用
process_data([1, 2, 3])
process_data([4, 5])
process_data([6])
print(f"总调用次数: {process_data.call_count}")
输出：
函数 process_data 被调用了 1 次
处理数据: [1, 2, 3]
函数 process_data 被调用了 2 次
处理数据: [4, 5]
函数 process_data 被调用了 3 次
处理数据: [6]
总调用次数: 3
带参数的类装饰器
class Retry:
"""重试机制的类装饰器"""
def __init__(self, max_retries=3, delay=1.0, exceptions=(Exception,)):
self.max_retries = max_retries
self.delay = delay
self.exceptions = exceptions
def __call__(self, func):
import time
import functools
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except self.exceptions as e:
last_exception = e
if attempt < self.max_retries:
print(f"第 {attempt + 1} 次尝试失败: {e}，等待 {self.delay} 秒后重试...")
time.sleep(self.delay)
else:
print(f"所有 {self.max_retries + 1} 次尝试均失败")
raise last_exception
return None
return wrapper
@Retry(max_retries=3, delay=0.5, exceptions=(ValueError, ConnectionError))
def unstable_operation():
import random
if random.random() < 0.3:
return "成功"
else:
raise ValueError("随机失败")
# 测试（可能需要多次运行才能看到所有情况）
try:
result = unstable_operation()
print(f"结果: {result}")
except ValueError as e:
print(f"最终错误: {e}")
装饰器的高级应用
1. 使用functools.wraps保留元数据
import functools
def logging_decorator(func):
@functools.wraps(func) # 保留原函数的名称、文档字符串等元数据
def wrapper(*args, **kwargs):
print(f"调用 {func.__name__}，参数: {args}, {kwargs}")
result = func(*args, **kwargs)
print(f"{func.__name__} 返回: {result}")
return result
return wrapper
@logging_decorator
def calculate_area(radius):
"""计算圆的面积"""
import math
return math.pi * radius ** 2
print(f"函数名: {calculate_area.__name__}")
print(f"文档字符串: {calculate_area.__doc__}")
print(f"模块: {calculate_area.__module__}")
2. 多个装饰器的组合使用
def bold(func):
def wrapper():
return f"<b>{func()}</b>"
return wrapper
def italic(func):
def wrapper():
return f"<i>{func()}</i>"
return wrapper
def underline(func):
def wrapper():
return f"<u>{func()}</u>"
return wrapper
@bold
@italic
@underline
def hello():
return "Hello World"
print(hello()) # 输出: <b><i><u>Hello World</u></i></b>
3. 类方法装饰器
class PaymentProcessor:
@staticmethod
def validate_amount(func):
def wrapper(self, amount):
if amount <= 0:
raise ValueError("金额必须大于0")
if amount > 1000000:
raise ValueError("金额超过限制")
return func(self, amount)
return wrapper
@validate_amount
def process_payment(self, amount):
print(f"处理支付: {amount} 元")
return f"支付成功: {amount} 元"
processor = PaymentProcessor()
print(processor.process_payment(100))
try:
processor.process_payment(-50)
except ValueError as e:
print(f"错误: {e}")
4. 基于类的装饰器支持实例方法
class AccessControl:
def __init__(self, required_role):
self.required_role = required_role
def __call__(self, func):
def wrapper(instance, *args, **kwargs):
# instance是类实例
user_role = getattr(instance, 'role', 'guest')
if user_role != self.required_role:
raise PermissionError(f"需要 {self.required_role} 权限，当前用户角色: {user_role}")
return func(instance, *args, **kwargs)
return wrapper
class User:
def __init__(self, name, role):
self.name = name
self.role = role
@AccessControl('admin')
def delete_user(self, user_id):
print(f"删除用户 {user_id}")
return True
def view_user(self, user_id):
print(f"查看用户 {user_id}")
return True
admin = User("管理员", "admin")
guest = User("访客", "guest")
admin.delete_user(123) # 成功
guest.view_user(456) # 成功
try:
guest.delete_user(789) # 失败
except PermissionError as e:
print(f"权限错误: {e}")
实际应用场景
1. 性能监控装饰器
import time
import functools
from typing import Callable, Any
def performance_monitor(threshold_ms: float = 100.0):
"""监控函数执行时间，超过阈值时发出警告"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
start_time = time.perf_counter()
result = func(*args, **kwargs)
end_time = time.perf_counter()
execution_time = (end_time - start_time) * 1000 # 转换为毫秒
if execution_time > threshold_ms:
print(f"⚠️ 性能警告: {func.__name__} 执行耗时 {execution_time:.2f}ms")
else:
print(f"✅ {func.__name__} 执行耗时 {execution_time:.2f}ms")
return result
return wrapper
return decorator
@performance_monitor(threshold_ms=50)
def slow_function():
time.sleep(0.06) # 60ms
return "慢函数完成"
@performance_monitor(threshold_ms=50)
def fast_function():
time.sleep(0.01) # 10ms
return "快函数完成"
slow_function()
fast_function()
2. 缓存装饰器（简化版LRU缓存）
import functools
from typing import Callable, Any
def lru_cache(max_size: int = 128):
"""简化版LRU缓存装饰器"""
def decorator(func: Callable) -> Callable:
cache = {}
order = [] # 记录访问顺序
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
# 将可变参数转换为可哈希的键
key = (args, tuple(sorted(kwargs.items())))
if key in cache:
# 更新访问顺序
order.remove(key)
order.append(key)
print(f"从缓存返回: {func.__name__}({args}, {kwargs})")
return cache[key]
result = func(*args, **kwargs)
# 缓存结果
if len(cache) >= max_size:
# 移除最旧的条目
oldest_key = order.pop(0)
del cache[oldest_key]
cache[key] = result
order.append(key)
print(f"计算并缓存: {func.__name__}({args}, {kwargs})")
return result
# 添加缓存统计方法
def cache_info():
return {
"cache_size": len(cache),
"max_size": max_size,
"hit_rate": f"{(len(order) - len(cache)) / len(order) * 100:.1f}%" if order else "0%"
}
wrapper.cache_info = cache_info
return wrapper
return decorator
@lru_cache(max_size=3)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
print("计算 fibonacci(5):")
result = fibonacci(5)
print(f"结果: {result}")
print(f"缓存信息: {fibonacci.cache_info()}")
print("\n再次计算 fibonacci(3):")
result2 = fibonacci(3)
print(f"结果: {result2}")
print(f"缓存信息: {fibonacci.cache_info()}")
3. 事务管理装饰器
import functools
from contextlib import contextmanager
class DatabaseConnection:
"""模拟数据库连接"""
def __init__(self):
self.connected = False
self.committed = False
def connect(self):
print("连接数据库...")
self.connected = True
def close(self):
print("关闭数据库连接")
self.connected = False
def commit(self):
if not self.connected:
raise RuntimeError("未连接数据库")
print("提交事务")
self.committed = True
def rollback(self):
if not self.connected:
raise RuntimeError("未连接数据库")
print("回滚事务")
def transactional(func):
"""事务管理装饰器"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
db = DatabaseConnection()
try:
db.connect()
result = func(db, *args, **kwargs)
db.commit()
return result
except Exception as e:
print(f"发生错误: {e}")
db.rollback()
raise
finally:
db.close()
return wrapper
@transactional
def transfer_money(db, from_user, to_user, amount):
print(f"从 {from_user} 转账 {amount} 元给 {to_user}")
if amount <= 0:
raise ValueError("转账金额必须大于0")
return f"转账成功: {amount} 元"
# 成功转账
try:
result = transfer_money("Alice", "Bob", 100)
print(result)
except Exception as e:
print(f"转账失败: {e}")
# 失败转账
try:
result = transfer_money("Charlie", "David", -50)
print(result)
except Exception as e:
print(f"转账失败: {e}")
4. API限流装饰器
import time
import functools
from collections import deque
from threading import Lock
class RateLimiter:
"""基于滑动窗口的限流器"""
def __init__(self, max_calls: int, period: float):
self.max_calls = max_calls
self.period = period
self.calls = deque()
self.lock = Lock()
def allow_request(self) -> bool:
with self.lock:
now = time.time()
# 移除过期的调用记录
while self.calls and self.calls[0] < now - self.period:
self.calls.popleft()
if len(self.calls) < self.max_calls:
self.calls.append(now)
return True
return False
def rate_limit(max_calls: int, period: float):
"""限流装饰器"""
def decorator(func):
limiter = RateLimiter(max_calls, period)
@functools.wraps(func)
def wrapper(*args, **kwargs):
if limiter.allow_request():
return func(*args, **kwargs)
else:
raise Exception(f"请求过于频繁，请等待 {period} 秒")
return wrapper
return decorator
@rate_limit(max_calls=3, period=5.0)
def api_call(endpoint):
print(f"调用API: {endpoint}")
return {"status": "success", "endpoint": endpoint}
# 测试限流
for i in range(5):
try:
result = api_call(f"/api/v1/resource/{i}")
print(f"第 {i+1} 次调用成功")
except Exception as e:
print(f"第 {i+1} 次调用失败: {e}")
time.sleep(0.5)
装饰器的最佳实践和注意事项
1. 使用functools.wraps
始终使用@functools.wraps来保留原函数的元数据，包括函数名、文档字符串、注解等。
import functools
def proper_decorator(func):
@functools.wraps(func) # ✅ 正确做法
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
def improper_decorator(func): # ❌ 错误做法
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
2. 保持装饰器的单一职责
# ✅ 好的做法：拆分为多个装饰器
@log_execution
@validate_input
@retry_on_failure
def complex_operation(data):
pass
# ❌ 不好的做法：所有功能混在一起
@everything_decorator
def complex_operation(data):
pass
3. 考虑装饰器的性能影响
import time
def heavy_computation_decorator(func):
"""这个装饰器有显著的性能开销"""
def wrapper(*args, **kwargs):
# 每次调用都进行复杂计算
start = time.time()
result = func(*args, **kwargs)
end = time.time()
if end - start > 0.1:
print(f"慢函数: {func.__name__}")
return result
return wrapper
# 对于频繁调用的小函数，考虑是否真的需要装饰器
4. 调试装饰器的技巧
def debug_decorator(func):
def wrapper(*args, **kwargs):
import traceback
print(f"=== 调试 {func.__name__} ===")
print(f"参数: args={args}, kwargs={kwargs}")
try:
result = func(*args, **kwargs)
print(f"返回: {result}")
return result
except Exception as e:
print(f"异常: {e}")
traceback.print_exc()
raise
finally:
print(f"=== 调试结束 ===\n")
return wrapper
@debug_decorator
def test_function(x):
return x * 2
test_function(5)
装饰器的替代方案
在某些情况下，装饰器可能不是最佳选择：
1. 上下文管理器
# 装饰器方式
@transactional
def process():
pass
# 上下文管理器方式（更灵活）
def process():
with transaction():
pass
2. 函数组合
# 装饰器
@log
@validate
def func():
pass
# 函数组合
def func():
return validate(log(original_func))()
总结
Python装饰器是一个强大而灵活的工具，它允许我们以声明式的方式为函数添加功能。通过本文的详细讲解和丰富的示例，我们涵盖了：
- 基础概念：装饰器的语法、工作原理和闭包机制
- 参数化装饰器：如何创建接受参数的装饰器
- 类装饰器：使用类实现更复杂的装饰逻辑
- 高级应用：性能监控、缓存、事务管理、限流等实际场景
- 最佳实践：使用functools.wraps、保持单一职责、性能考虑等
装饰器的核心优势在于其非侵入性，它让我们能够在不修改原有代码的情况下扩展功能，这在大型项目中尤其重要。然而，我们也需要注意装饰器的滥用可能导致代码难以理解和调试。
在实际开发中，建议：
- 优先使用标准库中的装饰器（如
@property、@staticmethod、@functools.lru_cache） - 为装饰器编写清晰的文档和类型注解
- 在性能敏感的场景下测试装饰器的开销
- 考虑使用类装饰器来维护状态
通过合理使用装饰器，我们可以写出更加优雅、可维护和可扩展的Python代码。