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

Python 装饰器高级用法与实战技巧

admin2周前 (03-23)Python22

装饰器(Decorator)是 Python 中最优雅的特性之一,但很多开发者只停留在使用 @staticmethod 或 @property 的层面。本文将深入探讨装饰器的高级用法,帮助你写出更优雅、更高效的代码。

一、装饰器基础回顾

装饰器本质上是一个接受函数作为参数,并返回一个新函数的高阶函数。它允许在不修改原函数代码的情况下,为函数添加额外的功能。

1.1 最简单的装饰器

def my_decorator(func):\n    def wrapper(*args, **kwargs):\n        print(f"📌 调用函数: {func.__name__}")\n        result = func(*args, **kwargs)\n        print(f"✅ 函数执行完毕: {func.__name__}")\n        return result\n    return wrapper\n\n@my_decorator\ndef greet(name):\n    return f"你好,{name}!"\n\nprint(greet("小豆包"))

输出:

📌 调用函数: greet\n✅ 函数执行完毕: greet\n你好,小豆包!

1.2 使用 functools.wraps 保留元数据

装饰器会覆盖原函数的元数据(如 __name__、__doc__),这在调试和文档生成时会有问题:

import functools\n\ndef proper_decorator(func):\n    @functools.wraps(func)  # 保留原函数的元数据\n    def wrapper(*args, **kwargs):\n        return func(*args, **kwargs)\n    return wrapper\n\n@proper_decorator\ndef calculate(x, y):\n    """执行加法运算"""\n    return x   y\n\nprint(calculate.__name__)  # 输出: calculate\nprint(calculate.__doc__)   # 输出: 执行加法运算

二、带参数的装饰器

当装饰器需要接受参数时,需要三层嵌套:装饰器工厂 → 装饰器 → 包装函数

2.1 可重复执行的装饰器

import functools\n\ndef repeat(times):\n    """创建一个重复执行函数多次的装饰器"""\n    def decorator(func):\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            results = []\n            for _ in range(times):\n                result = func(*args, **kwargs)\n                results.append(result)\n            return results[-1]  # 返回最后一次执行的结果\n        return wrapper\n    return decorator\n\n@repeat(times=3)\ndef expensive_calculation():\n    """模拟耗时计算"""\n    print("⏳ 执行计算中...")\n    return 42\n\nprint(expensive_calculation())

输出:

⏳ 执行计算中...\n⏳ 执行计算中...\n⏳ 执行计算中...\n42

2.2 可选参数的装饰器

def smart_timer(prefix="耗时"):\n    """可选参数的计时装饰器"""\n    def decorator(func):\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            import time\n            start = time.time()\n            result = func(*args, **kwargs)\n            end = time.time()\n            print(f"{prefix}: {end - start:.4f} 秒")\n            return result\n        return wrapper\n    return decorator\n\n# 使用自定义前缀\n@smart_timer(prefix="处理时间")\ndef process_data():\n    import time\n    time.sleep(0.1)\n    return "处理完成"\n\nprocess_data()

三、类装饰器

装饰器不仅限于函数,类也可以作为装饰器,这在需要维护状态时特别有用。

3.1 基础类装饰器

import functools\n\nclass CountCalls:\n    """统计函数调用次数的类装饰器"""\n    def __init__(self, func):\n        self.func = func\n        self.count = 0\n\n    def __call__(self, *args, **kwargs):\n        self.count  = 1\n        print(f"🔢 函数 {self.func.__name__} 已调用 {self.count} 次")\n        return self.func(*args, **kwargs)\n\n@CountCalls\ndef fetch_user(user_id):\n    return f"用户 {user_id} 的数据"\n\nfetch_user(1)\nfetch_user(2)\nfetch_user(3)

输出:

🔢 函数 fetch_user 已调用 1 次\n🔢 函数 fetch_user 已调用 2 次\n🔢 函数 fetch_user 已调用 3 次

3.2 带参数的类装饰器

class ValidateInput:\n    """输入验证装饰器"""\n    def __init__(self, min_value=None, max_value=None):\n        self.min_value = min_value\n        self.max_value = max_value\n\n    def __call__(self, func):\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            for arg in args:\n                if self.min_value is not None and arg < self.min_value:\n                    raise ValueError(f"参数 {arg} 小于最小值 {self.min_value}")\n                if self.max_value is not None and arg > self.max_value:\n                    raise ValueError(f"参数 {arg} 大于最大值 {self.max_value}")\n            return func(*args, **kwargs)\n        return wrapper\n\n@ValidateInput(min_value=0, max_value=100)\ndef calculate_score(points):\n    return f"得分: {points}"\n\nprint(calculate_score(85))\nprint(calculate_score(150))  # 这里会抛出异常

四、装饰器链

多个装饰器可以叠加使用,执行顺序从下往上

import functools\nimport time\n\ndef log_execution(func):\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        print(f"🔍 开始执行: {func.__name__}")\n        return func(*args, **kwargs)\n    return wrapper\n\ndef measure_time(func):\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        start = time.time()\n        result = func(*args, **kwargs)\n        end = time.time()\n        print(f"⏱️ 执行时间: {end - start:.2f} 秒")\n        return result\n    return wrapper\n\ndef cache_result(func):\n    """简单的缓存装饰器"""\n    cache = {}\n    @functools.wraps(func)\n    def wrapper(*args, **kwargs):\n        key = str(args)   str(kwargs)\n        if key not in cache:\n            cache[key] = func(*args, **kwargs)\n            print("💾 缓存未命中,执行计算")\n        else:\n            print("✅ 缓存命中")\n        return cache[key]\n    return wrapper\n\n@log_execution\n@measure_time\n@cache_result\ndef fibonacci(n):\n    """计算斐波那契数列"""\n    if n <= 1:\n        return n\n    return fibonacci(n-1)   fibonacci(n-2)\n\nprint(f"fibonacci(10) = {fibonacci(10)}")\nprint(f"fibonacci(10) = {fibonacci(10)}")  # 第二次会从缓存读取

五、实战应用场景

5.1 API 请求限流

import functools\nimport time\nfrom datetime import datetime\n\nclass RateLimiter:\n    """API 请求限流装饰器"""\n    def __init__(self, calls_per_minute=60):\n        self.calls_per_minute = calls_per_minute\n        self.call_times = []\n\n    def __call__(self, func):\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            current_time = time.time()\n            # 移除 1 分钟前的调用记录\n            self.call_times = [\n                t for t in self.call_times\n                if current_time - t < 60\n            ]\n            # 检查是否超过限制\n            if len(self.call_times) >= self.calls_per_minute_minute:\n                wait_time = 60 - (current_time - self.call_times[0])\n                print(f"⏳ 请求限流,等待 {wait_time:.1f} 秒...")\n                time.sleep(wait_time)\n            # 记录调用时间\n            self.call_times.append(current_time)\n            return func(*args, **kwargs)\n        return wrapper\n\n@RateLimiter(calls_per_minute=3)\ndef api_request(endpoint):\n    """模拟 API 请求"""\n    timestamp = datetime.now().strftime("%H:%M:%S")\n    print(f"📡 API 请求到 {endpoint} [{timestamp}]")\n    return {"status": "success"}\n\n# 测试限流\nfor i in range(5):\n    api_request(f"/api/data/{i}")

5.2 自动重试机制

import functools\nimport time\nimport random\n\ndef retry(max_attempts=3, delay=1, backoff=2):\n    """自动重试装饰器,支持指数退避"""\n    def decorator(func):\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            attempts = 0\n            current_delay = delay\n            while attempts < max_attempts:\n                try:\n                    return func(*args, **kwargs)\n                except Exception as e:\n                    attempts  = 1\n                    if attempts >= max_attempts:\n                        print(f"❌ 重试 {max_attempts} 次后失败: {e}")\n                        raise\n                    print(f"⚠️ 第 {attempts} 次失败: {e},{current_delay} 秒后重试...")\n                    time.sleep(current_delay)\n                    current_delay *= backoff\n        return wrapper\n    return decorator\n\n@retry(max_attempts=3, delay=1, backoff=2)\ndef unstable_service():\n    """模拟不稳定的服务"""\n    if random.random() < 0.7:\n        raise ConnectionError("网络连接失败")\n    return {"data": "成功获取数据"}\n\nprint(unstable_service())

5.3 异步装饰器

import functools\nimport asyncio\n\nasync def async_timer(func):\n    """异步函数计时装饰器"""\n    @functools.wraps(func)\n    async def wrapper(*args, **kwargs):\n        start = asyncio.get_event_loop().time()\n        result = await func(*args, **kwargs)\n        end = asyncio.get_event_loop().time()\n        print(f"⏱️ 异步函数 {func.__name__} 耗时: {end - start:.2f} 秒")\n        return result\n    return wrapper\n\n@async_timer\nasync def async_task():\n    """模拟异步任务"""\n    await asyncio.sleep(1)\n    return "任务完成"\n\n# 运行异步任务\nasyncio.run(async_task())

六、性能优化技巧

6.1 使用缓存避免重复计算

import functools\n\ndef memoize(max_size=128):\n    """带带大小限制的缓存装饰器"""\n    def decorator(func):\n        cache = {}\n        order = []  # 维护访问顺序\n\n        @functools.wraps(func)\n        def wrapper(*args, **kwargs):\n            key = str(args)   str(sorted(kwargs.items()))\n            if key in cache:\n                # 缓存命中\n                order.remove(key)\n                order.append(key)\n                return cache[key]\n            else:\n                # 缓存未命中\n                result = func(*args, **kwargs)\n                if len(cache) >= max_size:\n                    oldest_key = order.pop(0)\n                    del cache[oldest_key]\n                cache[key] = result\n                order.append(key)\n                return result\n        return wrapper\n    return decorator\n\n@memoize(max_size=3)\ndef heavy_computation(n):\n    """模拟重计算"""\n    print(f"🔥 计算中: {n}")\n    import time\n    time.sleep(0.1)\n    return n * n\n\nprint(heavy_computation(5))\nprint(heavy_computation(10))\nprint(heavy_computation(5))  {  # 从缓存读取\nprint(heavy_computation(15))\nprint(heavy_computation(10))  # 从缓存读取

6.2 延迟加载装饰器

import functools\n\ndef lazy_property(func):\n    """延迟加载属性,只在第一次访问时计算"""\n    attr_name = f"_lazy_{func.__name__}"\n\n    @property\n    @functools.wraps(func)\n    def wrapper(self):\n        if not hasattr(self, attr_name):\n            setattr(self, attr_name, func(self))\n        return getattr getattr(self, attr_name)\n    return wrapper\n\nclass DataProcessor:\n    def __init__(self, data):\n        self.data = data\n\n    @lazy_property\n    def processed_data(self):\n        print("🔄 开始处理数据...")\n        # 模拟耗时操作\n        import time\n        time.sleep(0.5)\n        return [x * 2 for x in self.data]\n\nprocessor = DataProcessor([1, 2, 3])\nprint("对象创建完成")\nprint(f"第一次访问: {processor.processed_data}")  # 触发计算\nprint(f"第二次访问: {processor.processed_data}")   # 使用缓存

七、最佳实践建议

  1. 始终使用 functools.wraps:保留原函数的元数据,便于调试
  2. 装饰器应该轻量级:不要在装饰器中执行耗时操作
  3. 考虑线程安全:如果装饰器维护状态,需要考虑并发访问
  4. 提供清晰的错误信息:装饰器捕获异常时,保留原始异常信息
  5. 文档化你的装饰器:为装饰器添加详细的 docstring

八、总结

Python 装饰器是一个强大的工具,它能够:

  • ✅ 横切关注点的分离(日志、缓存、限流)
  • ✅ 代码复用和模块化
  • ✅ 运行时行为修改
  • ✅ 优雅地增强函数功能

掌握装饰器的高级用法,将让你的 Python 代码更加优雅、高效和可维护。从简单的计时器到复杂的缓存系统,装饰器都能帮你简化代码结构,提升开发效率。

进一步学习:

  • 尝试编写一个上下文管理器装饰器
  • 探索装饰器在 Web 框架中的应用(如 Flask、Django)
  • 学习使用 wrapt 库处理复杂的装饰器场景

相关文章

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

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

Python 装饰器的实用技巧与高级用法

装饰器本质上是一个接受函数作为参数并返回新函数的高阶函数。理解这一点是掌握装饰器的关键。让我们从最简单的例子开始,逐步深入到复杂的应用场景。首先,我们需要理解函数在 Python 中是一等公民。这意味...

Python装饰器完全指南:从基础到高级应用

装饰器是 Python 中最强大也最容易被误解的特性之一。很多初学者听说过装饰器,但总是感觉云里雾里,不敢在实际项目中使用。本文从最基础的概念讲起,逐步深入到高级应用场景,通过大量原创示例代码帮助...

深入理解 Python 上下文管理器:从基础到高级应用

Python 的 with 语句和上下文管理器是每个开发者都应该掌握的高级技巧,但很多初学者对它的理解仅仅停留在文件操作层面。本文将深入讲解上下文管理器的原理、多种实现方式,以及在实际开发中的高级应用...

深入理解 Python 装饰器:从基础到高级的完整指南

什么是装饰器?装饰器本质上是一个 Python 函数,它可以让其他函数在不需要做任何代码变动的前提下增加额外功能。装饰器的返回值通常也是一个函数对象。这种设计模式遵循了"开放封闭原则"——对扩展开放,...

Python 装饰器完全指南:从入门到精通

什么是装饰器?装饰器本质上是一个 Python 函数,它可以让其他函数在不需要做任何代码变动的前提下增加额外功能。装饰器的返回值也是一个函数对象。在 Python 中,装饰器使用 @decorator...

发表评论

访客

看不清,换一张

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