当前位置:首页 > Python > 正文内容

Python 装饰器高级实战:从基础到精通的5个实用技巧

admin5小时前Python4

引言:为什么要深入掌握装饰器?

装饰器是 Python 中最优雅的元编程工具之一,它能在不修改原函数代码的情况下,动态地增加功能。很多开发者都知道如何使用 @timer 计时或 @cache 缓存,但装饰器的真正威力远不止于此。

当你需要在多个函数中添加日志、权限检查、重试逻辑、性能监控等横切关注点时,装饰器能让你避免代码重复,保持业务逻辑的纯净。本文将从实战角度出发,分享 5 个高级装饰器技巧,每个都配有可直接运行的代码示例。

技巧一:带参数的装饰器工厂

基础装饰器只能接受被装饰的函数,但实际项目中,我们经常需要给装饰器传递参数。比如创建一个可配置的重试装饰器,可以指定最大重试次数和延迟时间。

import time
from functools import wraps

def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
    """带参数的装饰器工厂函数"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts:
                        print(f"尝试 {attempt}/{max_attempts} 失败: {e}, {delay}秒后重试...")
                        time.sleep(delay)
            raise Exception(f"所有 {max_attempts} 次尝试都失败了") from last_exception
        return wrapper
    return decorator

# 使用示例
@retry(max_attempts=3, delay=2, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络连接失败")
    return f"从 {url} 获取的数据"

# 测试
try:
    result = fetch_data("https://api.example.com")
    print(f"成功: {result}")
except Exception as e:
    print(f"最终失败: {e}")

这里的 key 理解:retry() 返回真正的装饰器,decorator(func) 返回包装函数。这种三层嵌套结构是带参数装饰器的标准模式。

技巧二:类装饰器 - 更强大的状态管理

函数装饰器适合简单场景,但当装饰器需要维护复杂状态或提供额外方法时,类装饰器是更好的选择。类装饰器可以实现类似中间件的链式调用。

class AccessLogger:
    """带状态管理的类装饰器"""
    
    def __init__(self, log_file="access.log"):
        self.log_file = log_file
        self.call_count = {}  # 记录每个函数的调用次数
    
    def __call__(self, func):
        def wrapper(*args, **kwargs):
            # 记录调用
            func_name = func.__name__
            self.call_count[func_name] = self.call_count.get(func_name, 0) + 1
            
            # 记录日志
            timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
            log_entry = f"[{timestamp}] {func_name} 被调用 (第{self.call_count[func_name]}次)\n"
            
            with open(self.log_file, "a") as f:
                f.write(log_entry)
            
            print(f"📊 {func_name} 累计调用: {self.call_count[func_name]} 次")
            return func(*args, **kwargs)
        return wrapper
    
    def get_stats(self):
        """获取统计信息"""
        return self.call_count.copy()

# 使用示例
logger = AccessLogger()

@logger
def process_user_data(user_id):
    return f"处理用户 {user_id} 的数据"

@logger
def send_notification(message):
    return f"发送通知: {message}"

# 测试
process_user_data(1001)
process_user_data(1002)
send_notification("欢迎注册")
send_notification("密码重置")
print(f"\n统计: {logger.get_stats()}")

类装饰器的优势在于可以存储状态(如调用计数、日志信息),并且可以提供额外方法(如 get_stats())来查询这些状态。

技巧三:装饰器堆叠顺序 - 谁先执行?

当多个装饰器堆叠使用时,理解执行顺序至关重要。装饰器的执行顺序是"从下到上注册,从上到下执行"。

def auth_check(required_role="user"):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"🔐 权限检查: 需要 {required_role}权限")
            return func(*args, **kwargs)
        return wrapper
    return decorator

def rate_limit(max_calls=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"⚡ 限流检查: 最多 {max_calls} 次/分钟")
            return func(*args, **kwargs)
        return wrapper
    return decorator

def log_performance(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        print(f"📏 性能监控: 开始计时")
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        print(f"📏 性能监控: 耗时 {elapsed:.4f} 秒")
        return result
    return wrapper

# 装饰器堆叠 - 注意顺序!
@rate_limit(max_calls=10)
@auth_check(required_role="admin")
@log_performance
def sensitive_operation(data):
    print(f"执行敏感操作: {data}")
    return "操作完成"

# 执行
result = sensitive_operation("删除重要文件")

print("\n--- 执行顺序解析 ---")
print("1. rate_limit (最先执行,最外层)")
print("2. auth_check (第二层)")
print("3. log_performance (第三层)")
print("4. sensitive_operation (原函数)")
print("\n实际调用链: rate_limit → auth_check → log_performance → sensitive_operation")

实战建议:将最通用的装饰器(如日志、性能监控)放在最下面,将最具体的(如权限检查)放在最上面。这样可以让后续装饰器基于前面的装饰器结果进行决策。

技巧四:异步装饰器 - 适配 async/await

现代 Python 项目中大量使用异步编程。装饰器也需要正确支持异步函数,否则会导致 RuntimeWarning 或协程泄漏。

import asyncio
from functools import wraps

def async_cache(ttl=60):
    """支持异步函数的缓存装饰器"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 生成缓存键
            cache_key = (func.__name__, args, frozenset(kwargs.items()))
            
            # 检查缓存
            if cache_key in cache:
                cached_time, result = cache[cache_key]
                if time.time() - cached_time < ttl:
                    print(f"🎯 缓存命中: {func.__name__}")
                    return result
            
            # 执行原函数
            print(f"🔄 执行: {func.__name__}")
            result = await func(*args, **kwargs)
            
            # 存入缓存
            cache[cache_key] = (time.time(), result)
            return result
        return wrapper
    return decorator

def async_timeout(seconds=10):
    """异步函数超时控制装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await asyncio.wait_for(func(*args, **kwargs), timeout=seconds)
            except asyncio.TimeoutError:
                raise TimeoutError(f"{func.__name__} 执行超时 ({seconds}秒)")
        return wrapper
    return decorator

# 使用示例
@async_cache(ttl=5)
@async_timeout(seconds=3)
async def fetch_user_data(user_id):
    print(f"查询用户 {user_id} 数据...")
    await asyncio.sleep(1)  # 模拟网络请求
    return {"id": user_id, "name": f"User{user_id}", "email": f"user{user_id}@example.com"}

async def main():
    # 第一次调用 - 会执行
    print("--- 第一次调用 ---")
    data1 = await fetch_user_data(1001)
    print(f"结果: {data1}")
    
    # 第二次调用 - 命中缓存
    print("\n--- 第二次调用 (缓存) ---")
    data2 = await fetch_user_data(1001)
    print(f"结果: {data2}")

# 运行测试
asyncio.run(main())

关键点:async def wrapperawait func(*args, **kwargs)。注意 asyncio.wait_for() 可以优雅地处理超时,避免长时间挂起。

技巧五:装饰器性能优化 - 避免常见陷阱

装饰器虽然优雅,但使用不当会影响性能。以下是几个优化技巧:

1. 始终使用 functools.wraps

from functools import wraps

# ❌ 错误做法
def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# ✅ 正确做法
def good_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# 测试差异
@bad_decorator
def example_func():
    """这是一个示例函数"""
    pass

print(f"❌ 错误做法: {example_func.__name__}, 文档: {example_func.__doc__}")
# 输出: wrapper, None

@good_decorator
def example_func2():
    """这是一个示例函数"""
    pass

print(f"✅ 正确做法: {example_func2.__name__}, 文档: {example_func2.__doc__}")
# 输出: example_func2, 这是一个示例函数

2. 避免在装饰器中创建昂贵对象

# ❌ 每次调用都创建数据库连接
def bad_db_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        db = create_expensive_db_connection()  # 严重性能问题!
        result = func(*args, **kwargs, db=db)
        db.close()
        return result
    return wrapper

# ✅ 使用闭包或类装饰器缓存连接
class DBConnectionPool:
    _instance = None
    
    @classmethod
    def get_connection(cls):
        if cls._instance is None:
            cls._instance = create_expensive_db_connection()
        return cls._instance

def good_db_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        db = DBConnectionPool.get_connection()  # 复用连接
        return func(*args, **kwargs, db=db)
    return wrapper

3. 使用 __qualname__ 而非 __name__ 处理类方法

class MyClass:
    @good_decorator
    def method(self):
        pass

# __name__ 在类方法中可能不准确
print(f"__name__: {MyClass.method.__name__}")
print(f"__qualname__: {MyClass.method.__qualname__}")

总结:装饰器的最佳实践清单

使用 functools.wraps:保留原函数的元信息(名称、文档字符串等)
支持 *args, **kwargs:让装饰器更通用
考虑异步支持:如果项目使用 asyncio,确保装饰器兼容
合理控制堆叠顺序:通用装饰器在下,具体装饰器在上
避免重复计算:使用闭包或类属性缓存昂贵对象
提供清晰文档:装饰器应该有良好的 docstring 和示例

装饰器是提升 Python 代码质量和可维护性的利器。掌握这些高级技巧后,你可以在不修改业务代码的情况下,优雅地添加日志、缓存、权限检查、性能监控等功能。记住:装饰器应该让代码更清晰,而不是更复杂。

延伸阅读:如果想进一步探索,可以研究 Python 的 contextlib 模块(上下文管理器)和 __getattr____getattribute__ 等魔术方法,它们与装饰器一样,都是 Python 元编程的强大工具。

相关文章

[Python 教程] OpenCV-Python 入门:图像处理基础详解

OpenCV-Python 入门:图像处理基础详解OpenCV 是一个跨平台计算机视觉库,轻量级且高效,支持 Python 接口。本文将系统介绍 OpenCV 的核心概念和基础操作。一、OpenCV...

[Python 教程] OpenCV 实战:图像与视频文件处理

OpenCV 实战:图像与视频文件处理本文详细介绍如何使用 OpenCV 处理图像和视频文件,包括读取、显示、保存等操作。一、图像文件操作1.1 读取图像import cv2 #&nb...

[Python 教程] Pandas 数据分析实战

Pandas 数据分析实战 Pandas 是 Python 数据分析的核心库,提供 DataFrame 和 Series 数据结构。本文介绍 Pandas 的实用技巧。 一、创建 DataFrame...

[Python 教程] Matplotlib 数据可视化教程

Matplotlib 数据可视化教程 Matplotlib 是 Python 最常用的绘图库。本文介绍常用图表的绘制方法。 一、基础设置 import matplotlib.pyplot as pl...

[Python 教程] Python 网络请求与爬虫基础

Python 网络请求与爬虫基础 requests 是 Python 最常用的 HTTP 库。本文介绍网络请求和爬虫的基础知识。 一、基础请求 import requests # GET 请求 r...

Python 装饰器实用技巧:从入门到精通

装饰器是 Python 最强大的特性之一,但也是很多开发者感到困惑的概念。简单来说,装饰器是一个函数,它接受另一个函数作为输入,并返回一个新的函数。使用装饰器,你可以在不修改原函数代码的情况下,为其添...

发表评论

访客

看不清,换一张

◎欢迎参与讨论,请在这里发表您的看法和观点。