Python代码不够高效！看装饰器如何改变游戏规则？

2024-11-14

现代软件开发中，代码效率不仅关系到程序的运行速度，更影响着开发效率和代码的可维护性。Python作为一门优雅的编程语言，提供了许多强大的特性来提升代码质量，而装饰器（Decorator）就是其中最强大的特性之一。

1.1 什么是装饰器

装饰器是Python中的一个重要概念，它允许我们以非侵入式的方式修改或增强函数和类的行为。简单来说，装饰器就像是一个包装器，可以让我们在不改变原有代码的情况下，为其添加新的功能。

让我们看一个简单的例子：import time

def measure_time(func):
def wrapper():
start = time.time()
result = func()
print(f"函数执行时间: {time.time() - start}秒")
return result
return wrapper

@measure_time
def slow_function():
time.sleep(1)
return "完成"

# 使用装饰器
result = slow_function()
# 输出: 函数执行时间: 1.001秒

1.2 为什么需要装饰器

装饰器能帮助我们：

实现代码复用，避免重复编写相似的功能
提供更清晰的代码结构
实现横切关注点（如日志、性能监控等）
使代码更具可维护性和可读性

1.3 装饰器在Python生态中的应用

在很多流行的Python框架中，装饰器被广泛使用：# Flask中的路由装饰器
@app.route('/hello')
def hello():
return 'Hello, World!'

# Django中的视图装饰器
@login_required
def profile(request):
return render(request, 'profile.html')

# FastAPI中的路由装饰器
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}

2. 装饰器基础

2.1 装饰器的语法和工作原理

装饰器的基本语法如下：from functools import wraps

def basic_decorator(func):
@wraps(func) # 保持原函数的元数据
def wrapper(*args, **kwargs):
# 在函数执行前的操作
print("函数执行前")

result = func(*args, **kwargs) # 执行原函数

# 在函数执行后的操作
print("函数执行后")
return result
return wrapper

# 使用装饰器
@basic_decorator
def greet(name):
return f"Hello, {name}!"

2.1.1 装饰器的执行流程

# 装饰器执行流程
"""
1. 定义装饰器
@decorator
def function():
pass

2. Python解释器执行过程
step 1: decorator = decorator_factory(args) # 如果是带参数的装饰器
step 2: function = decorator(function)
step 3: function(*args, **kwargs) # 实际调用时
"""当Python解释器遇到装饰器时，会按照以下步骤执行：

1. 如果是带参数的装饰器，首先执行装饰器工厂函数，获取实际的装饰器
2. 将被装饰的函数作为参数传递给装饰器
3. 用装饰器返回的新函数替换原始函数
4. 在实际调用时，执行包装后的函数

例如，对于带参数的装饰器：@decorator(arg1, arg2)
def function():
pass

# 等价于
function = decorator(arg1, arg2)(function)

2.2 装饰器的类型

2.2.1 函数装饰器

def timing_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"函数 {func.__name__} 执行时间：{end_time - start_time}秒")
return result
return wrapper

@timing_decorator
def example_function():
time.sleep(1)

2.2.2 类装饰器

class TimingDecorator:
def __init__(self, func):
self.func = func

def __call__(self, *args, **kwargs):
start_time = time.time()
result = self.func(*args, **kwargs)
end_time = time.time()
print(f"函数 {self.func.__name__} 执行时间：{end_time - start_time}秒")
return result

def __set_name__(self, owner, name):
self.__name__ = name

@TimingDecorator
def example_class_decorator():
time.sleep(1)

2.3 装饰器链

装饰器可以叠加使用，从下往上依次执行："""
装饰器链执行顺序示意图：

@decorator1
@decorator2
@decorator3
def function():
pass

执行顺序：
1. decorator3(function) -> wrapped_3
2. decorator2(wrapped_3) -> wrapped_2
3. decorator1(wrapped_2) -> wrapped_1

调用时：wrapped_1() -> wrapped_2() -> wrapped_3() -> function()
"""

def decorator1(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("decorator1 开始")
result = func(*args, **kwargs)
print("decorator1 结束")
return result
return wrapper

def decorator2(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("decorator2 开始")
result = func(*args, **kwargs)
print("decorator2 结束")
return result
return wrapper

@decorator1 # 后执行
@decorator2 # 先执行
def example():
print("函数执行")

# 执行结果：
# decorator1 开始
# decorator2 开始
# 函数执行
# decorator2 结束
# decorator1 结束

3. 装饰器的高级特性

3.1 带参数的装饰器

def repeat(times=3):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for _ in range(times):
result = func(*args, **kwargs)
return result
return wrapper
return decorator

@repeat(times=2)
def greet(name):
print(f"Hello, {name}!")

3.2 保持函数元数据

使用functools.wraps保持原函数的元数据：from functools import wraps

def preserve_metadata(func):
@wraps(func) # 保持原函数的元数据
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper

@preserve_metadata
def example():
"""这是示例函数的文档字符串"""
pass

print(example.__name__) # 输出: example
print(example.__doc__) # 输出: 这是示例函数的文档字符串

4. 常见应用场景

4.1 统一的日志和错误处理

结合日志记录和错误处理的装饰器：import logging
from functools import wraps

def error_logging_handler(func):
"""
统一处理函数的日志记录和错误处理
- 记录函数的调用信息
- 记录函数的执行结果
- 捕获并记录异常信息
"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
# 记录函数调用
logging.info(f"调用函数: {func.__name__}, 参数: args={args}, kwargs={kwargs}")

# 执行函数
result = func(*args, **kwargs)

# 记录成功信息
logging.info(f"函数 {func.__name__} 执行成功, 返回值: {result}")
return result

except Exception as e:
# 记录错误信息
logging.error(f"函数 {func.__name__} 执行失败: {str(e)}", exc_info=True)
raise
return wrapper

# 使用示例
@error_logging_handler
def divide(a, b):
return a / b

# 测试正常情况
result = divide(10, 2) # 正常执行，记录信息
# 测试异常情况
result = divide(10, 0) # 触发异常，记录错误

4.2 性能监控和分析

完整的性能监控装饰器：import time
import psutil
import functools
from typing import Optional

def measure_performance(threshold_ms: Optional[float] = None):
"""
性能监控装饰器
:param threshold_ms: 性能警告阈值（毫秒）
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 记录开始状态
start_time = time.perf_counter()
start_memory = psutil.Process().memory_info().rss

# 执行函数
result = func(*args, **kwargs)

# 计算性能指标
end_time = time.perf_counter()
end_memory = psutil.Process().memory_info().rss
execution_time = (end_time - start_time) * 1000 # 转换为毫秒
memory_used = (end_memory - start_memory) / 1024 / 1024 # 转换为MB

# 记录性能数据
performance_data = {
"function": func.__name__,
"execution_time_ms": execution_time,
"memory_used_mb": memory_used
}

# 检查是否超过阈值
if threshold_ms and execution_time > threshold_ms:
logging.warning(
f"性能警告: 函数 {func.__name__} 执行时间 {execution_time:.2f}ms "
f"超过阈值 {threshold_ms}ms"
)

# 输出性能报告
print(f"性能报告 - {func.__name__}:")
print(f"执行时间: {execution_time:.2f}ms")
print(f"内存使用: {memory_used:.2f}MB")

return result
return wrapper
return decorator

# 使用示例
@measure_performance(threshold_ms=100)
def process_data(data: list):
time.sleep(0.2) # 模拟耗时操作
return sum(data)

4.3 缓存机制

实现带过期时间的缓存装饰器：import time
from functools import wraps
from typing import Optional
from collections import OrderedDict

class TimedCache:
def __init__(self, max_size: int = 100, expire_seconds: Optional[float] = None):
"""
初始化缓存
:param max_size: 最大缓存条目数
:param expire_seconds: 缓存过期时间（秒）
"""
self.max_size = max_size
self.expire_seconds = expire_seconds
self.cache = OrderedDict()

def get(self, key):
if key not in self.cache:
return None

value, timestamp = self.cache[key]

# 检查是否过期
if self.expire_seconds and time.time() - timestamp > self.expire_seconds:
del self.cache[key]
return None

# 更新访问顺序
self.cache.move_to_end(key)
return value

def set(self, key, value):
# 检查容量
if len(self.cache) >= self.max_size:
self.cache.popitem(last=False) # 删除最早的项

# 存储值和时间戳
self.cache[key] = (value, time.time())
self.cache.move_to_end(key)

def cached(max_size: int = 100, expire_seconds: Optional[float] = None):
"""
缓存装饰器
:param max_size: 最大缓存条目数
:param expire_seconds: 缓存过期时间（秒）
"""
cache = TimedCache(max_size, expire_seconds)

def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存键
key = str((func.__name__, args, frozenset(kwargs.items())))

# 尝试获取缓存
cached_value = cache.get(key)
if cached_value is not None:
print(f"缓存命中: {func.__name__}")
return cached_value

# 计算新值
result = func(*args, **kwargs)
cache.set(key, result)
print(f"缓存更新: {func.__name__}")
return result
return wrapper
return decorator

# 使用示例
@cached(max_size=100, expire_seconds=60)
def fibonacci(n: int) -> int:
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)

4.4 访问控制和认证

实现基于角色的访问控制装饰器：from enum import Enum
from typing import List, Union
from functools import wraps

class Role(Enum):
GUEST = "guest"
USER = "user"
ADMIN = "admin"

class User:
def __init__(self, user_id: int, roles: List[Role]):
self.user_id = user_id
self.roles = roles

def require_roles(required_roles: Union[Role, List[Role]]):
"""
角色验证装饰器
:param required_roles: 需要的角色（单个角色或角色列表）
"""
if isinstance(required_roles, Role):
required_roles = [required_roles]

def decorator(func):
@wraps(func)
def wrapper(user: User, *args, **kwargs):
# 验证用户角色
if not any(role in user.roles for role in required_roles):
raise PermissionError(
f"需要角色 {[role.value for role in required_roles]} 中的任意一个"
)
return func(user, *args, **kwargs)
return wrapper
return decorator

# 使用示例
@require_roles([Role.ADMIN])
def delete_user(current_user: User, user_id: int):
print(f"管理员 {current_user.user_id} 删除了用户 {user_id}")

@require_roles([Role.USER, Role.ADMIN])
def edit_profile(current_user: User, user_id: int, data: dict):
print(f"用户 {current_user.user_id} 编辑了资料")

4.5 数据库连接管理

from contextlib import contextmanager
import pymysql
from typing import Generator

@contextmanager
def database_connection() -> Generator[pymysql.Connection, None, None]:
"""数据库连接上下文管理器"""
conn = pymysql.connect(
host='localhost',
user='user',
password='password',
database='db_name'
)
try:
yield conn
finally:
conn.close()

def with_db_connection(func):
"""数据库连接装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
with database_connection() as conn:
return func(conn,*args,**kwargs)
return wrapper

4.6 装饰器工厂

class DecoratorFactory:
"""装饰器工厂类"""
@staticmethod
def create_logging_decorator(log_level: str = 'INFO'):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
logging.log(getattr(logging, log_level), f"调用函数: {func.__name__}")
return func(*args,**kwargs)
return wrapper
return decorator

@staticmethod
def create_timing_decorator(threshold_ms: float = None):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
duration = (time.perf_counter() - start) * 1000
if threshold_ms and duration > threshold_ms:
logging.warning(f"函数 {func.__name__} 执行时间超过阈值")
return result
return wrapper
return decorator

5. 实战案例

5.1 异步任务装饰器

import asyncio
from functools import wraps
from typing import Callable, Any

def async_task(func: Callable[..., Any]):
"""
将同步函数转换为异步任务的装饰器
"""
@wraps(func)
async def wrapper(*args, **kwargs):
# 在事件循环中运行同步函数
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, lambda: func(*args, **kwargs))
return wrapper

# 使用示例
@async_task
def cpu_intensive_task(n: int):
result = 0
for i in range(n):
result += i
return result

async def main():
# 异步执行CPU密集型任务
result = await cpu_intensive_task(1000000)
print(f"计算结果: {result}")

5.2 API响应格式化装饰器

from functools import wraps
from typing import Any, Dict
from datetime import datetime

def format_response(func):
"""
统一API响应格式的装饰器
"""
@wraps(func)
def wrapper(*args, **kwargs) -> Dict[str, Any]:
try:
result = func(*args, **kwargs)
return {
"success": True,
"data": result,
"timestamp": datetime.now().isoformat(),
"error": None
}
except Exception as e:
return {
"success": False,
"data": None,
"timestamp": datetime.now().isoformat(),
"error": {
"type": type(e).__name__,
"message": str(e)
}
}
return wrapper

# 使用示例
@format_response
def get_user_info(user_id: int):
if user_id < 0:
raise ValueError("用户ID不能为负数")
return {"id": user_id, "name": "测试用户"}

5.3 参数验证装饰器

from functools import wraps
from typing import Any, Callable, Dict, Type

def validate_params(**validators: Dict[str, Callable[[Any], bool]]):
"""
参数验证装饰器
:param validators: 参数名和对应的验证函数
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 获取函数参数名
func_params = func.__code__.co_varnames[:func.__code__.co_argcount]

# 合并位置参数和关键字参数
all_args = dict(zip(func_params, args))
all_args.update(kwargs)

# 验证参数
for param_name, validator in validators.items():
if param_name in all_args:
value = all_args[param_name]
if not validator(value):
raise ValueError(f"参数 {param_name} 验证失败: {value}")

return func(*args, **kwargs)
return wrapper
return decorator

# 验证函数
def is_positive(x: int) -> bool:
return isinstance(x, int) and x > 0

def is_valid_name(name: str) -> bool:
return isinstance(name, str) and 2 <= len(name) <= 50

# 使用示例
@validate_params(
age=is_positive,
name=is_valid_name
)
def register_user(name: str, age: int):
return f"注册用户: {name}, 年龄: {age}"

6. 装饰器的高级技巧

6.1 类方法装饰器

from functools import wraps
from typing import Type, TypeVar

T = TypeVar('T')

def class_method_decorator(method):
"""
类方法装饰器，可以访问类和实例属性
"""
@wraps(method)
def wrapper(cls_or_self, *args, **kwargs):
if isinstance(cls_or_self, type):
# 处理类方法调用
print(f"类方法调用: {cls_or_self.__name__}")
else:
# 处理实例方法调用
print(f"实例方法调用: {cls_or_self.__class__.__name__}")
return method(cls_or_self, *args, **kwargs)
return wrapper

class MyClass:
@class_method_decorator
@classmethod
def class_method(cls):
return "类方法"

@class_method_decorator
def instance_method(self):
return "实例方法"

6.2 异步装饰器

import asyncio
from functools import wraps
from typing import Callable, Awaitable, TypeVar

T = TypeVar('T')

def async_retry(
max_attempts: int = 3,
delay: float = 1.0
) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
"""
异步重试装饰器
:param max_attempts: 最大重试次数
:param delay: 重试间隔（秒）
"""
def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
@wraps(func)
async def wrapper(*args, **kwargs) -> T:
last_exception = None

for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_attempts - 1:
await asyncio.sleep(delay)
else:
raise last_exception

raise last_exception # 类型检查器需要这行

return wrapper
return decorator

# 使用示例
@async_retry(max_attempts=3, delay=1.0)
async def fetch_data(url: str) -> dict:
# 模拟异步网络请求
await asyncio.sleep(0.1)
if random.random() < 0.5: # 模拟随机失败
raise ConnectionError("网络错误")
return {"data": "success"}

7. 性能考虑

7.1 装饰器对性能的影响

装饰器虽然能提供很多便利，但也会带来一定的性能开销：

1. 函数调用开销

# 每次函数调用都会增加一层包装
def simple_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs) # 额外的函数调用层级
return wrapper

1. 内存占用

# 每个装饰器都会创建新的闭包对象
def decorator_with_state(func):
state = {} # 这个字典会一直存在于内存中
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper

7.2 性能优化技巧

1. 使用__slots__优化内存使用：

class OptimizedDecorator:
__slots__ = ['func'] # 限制实例属性，减少内存使用

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

def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)

1. 使用缓存减少重复计算：

from functools import lru_cache

@lru_cache(maxsize=None)
def expensive_decorator(func):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper

1. 避免不必要的装饰器嵌套：

# 不推荐
@decorator1
@decorator2
@decorator3
def function():
pass

# 推荐：合并装饰器功能
@combined_decorator
def function():
pass

7.3 性能测试示例

import time
import cProfile

def measure_decorator_overhead():
# 基准测试：无装饰器
def raw_function():
return sum(range(1000))

# 带装饰器的函数
@simple_decorator
def decorated_function():
return sum(range(1000))

# 性能测试
iterations = 100000

# 测试原始函数
start = time.perf_counter()
for _ in range(iterations):
raw_function()
raw_time = time.perf_counter() - start

# 测试装饰器函数
start = time.perf_counter()
for _ in range(iterations):
decorated_function()
decorated_time = time.perf_counter() - start

print(f"原始函数执行时间: {raw_time:.4f}秒")
print(f"装饰器函数执行时间: {decorated_time:.4f}秒")
print(f"装饰器开销: {((decorated_time - raw_time) / raw_time * 100):.2f}%")

8. 最佳实践

8.1 装饰器命名规范

# 好的命名示例
def validate_input(func):
pass

def cache_result(func):
pass

def require_authentication(func):
pass

# 避免的命名
def dec(func): # 太简短
pass

def my_decorator(func): # 不够具体
pass

8.2 文档化装饰器

def retry_on_failure(max_attempts=3, delay=1):
"""
在函数执行失败时自动重试的装饰器。

参数:
max_attempts (int): 最大重试次数
delay (int): 重试间隔时间（秒）

返回:
callable: 装饰后的函数

示例:
@retry_on_failure(max_attempts=3, delay=2)
def unstable_function():
pass
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
time.sleep(delay)
return wrapper
return decorator

8.3 调试装饰器

def debug_decorator(func):
"""用于调试的装饰器，打印函数的调用信息"""
@wraps(func)
def wrapper(*args, **kwargs):
# 打印函数调用信息
print(f"\n[DEBUG] 正在调用: {func.__name__}")
print(f"[DEBUG] 参数: {args}, {kwargs}")

# 记录开始时间
start_time = time.perf_counter()

try:
# 执行函数
result = func(*args, **kwargs)

# 打印返回值
print(f"[DEBUG] 返回值: {result}")
return result
except Exception as e:
# 打印异常信息
print(f"[DEBUG] 异常: {type(e).__name__}: {str(e)}")
raise
finally:
# 打印执行时间
end_time = time.perf_counter()
print(f"[DEBUG] 执行时间: {end_time - start_time:.4f}秒")

return wrapper

8.4 常见陷阱和解决方案

1. 装饰器顺序问题：

# 装饰器的执行顺序是从下到上的
@decorator1 # 第二个执行
@decorator2 # 第一个执行
def function():
pass

# 解决方案：注意装饰器的依赖关系，合理安排顺序

1. 参数传递问题：

def decorator_with_args(arg1, arg2):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 使用 arg1 和 arg2
return func(*args, **kwargs)
return wrapper
return decorator