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

Python 装饰器深度解析:从原理到实战

admin3周前 (03-23)Python24

装饰器是 Python 中最优雅的语言特性之一,它让我们能够在不修改原函数代码的情况下,动态地增强函数的功能。本文将深入讲解装饰器的底层原理、多种高级用法,并通过实际案例展示如何在项目中应用装饰器。

一、装饰器的本质:闭包与高阶函数

从本质上说,装饰器就是一个接收函数作为参数,并返回一个新函数的高阶函数。它的实现依赖于闭包这一机制。

def timer_decorator(func):
    """记录函数执行时间的装饰器"""
    import time
    import functools

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f"{func.__name__} 执行耗时: {end - start:.4f} 秒")
        return result
    return wrapper


@timer_decorator
def calculate_sum(n):
    """计算 1 到 n 的和"""
    total = sum(range(n + 1))
    return total


# 调用函数
result = calculate_sum(1000000)
print(f"计算结果: {result}")

这里使用了 functools.wraps 来保留原函数的元信息(如 __name____doc__ 等),这是一个重要的最佳实践。

二、带参数的装饰器工厂

当我们需要为装饰器传递配置参数时,可以使用装饰器工厂模式。这实际上是一个返回装饰器的函数。

def retry(max_attempts=3, delay=1):
    """失败重试装饰器工厂"""
    import time
    import functools

    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import random
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"第 {attempt + 1} 次尝试失败: {e},{delay} 秒后重试...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator


@retry(max_attempts=3, delay=0.5)
def unreliable_api_call():
    """模拟可能失败的 API 调用"""
    import random
    if random.random() < 0.7:  # 70% 概率失败
        raise ConnectionError("网络连接失败")
    return {"status": "success", "data": "API 返回数据"}


# 调用函数
try:
    result = unreliable_api_call()
    print(f"API 调用成功: {result}")
except Exception as e:
    print(f"API 调用最终失败: {e}")

三、类装饰器

除了函数装饰器,Python 还支持类装饰器。类装饰器通过实现 __call__ 方法,使类的实例可以像函数一样被调用。

class CacheDecorator:
    """缓存装饰器类"""

    def __init__(self, func):
        self.func = func
        self.cache = {}

    def __call__(self, *args, **kwargs):
        # 使用参数作为缓存键
        cache_key = (args, frozenset(kwargs.items()))
        if cache_key not in self.cache:
            print(f"计算 {self.func.__name__}({args[0]})...")
            self.cache[cache_key] = self.func(*args, **kwargs)
        else:
            print(f"从缓存获取 {self.func.__name__}({args[0]})...")
        return self.cache[cache_key]


@CacheDecorator
def fibonacci(n):
    """计算斐波那契数列"""
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)


# 调用函数
print(f"fibonacci(10) = {fibonacci(10)}")  # 会计算
print(f"fibonacci(10) = {fibonacci(10)}")  # 从缓存获取
print(f"fibonacci(15) = {fibonacci(15)}")  # 会计算(部分结果已缓存)

四、多个装饰器的链式调用

Python 允许为一个函数应用多个装饰器,装饰器的执行顺序是从下到上(靠近函数的装饰器先执行)。

def log_before(func):
    """执行前记录日志"""
    def wrapper(*args, **kwargs):
        print(f">>> 即将执行 {func.__name__},参数: {args}, {kwargs}")
        return func(*args, **kwargs)
    return wrapper


def log_after(func):
    """执行后记录日志"""
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        print(f"<<< {func.__name__} 执行完成,返回值: {result}")
        return result
    return wrapper


@log_after
@log_before
def multiply(a, b):
    """两个数相乘"""
    return a * b


# 调用函数
result = multiply(5, 3)
print(f"最终结果: {result}")

五、实战案例:权限验证装饰器

在实际开发中,装饰器常用于权限验证、日志记录、性能监控等场景。下面实现一个实用的权限验证装饰器。

class PermissionDeniedError(Exception):
    """权限不足异常"""
    pass


def require_role(*allowed_roles):
    """权限验证装饰器"""
    def decorator(func):
        def wrapper(user, *args, **kwargs):
            if user.get('role') not in allowed_roles:
                raise PermissionDeniedError(
                    f"用户 {user.get('name')} 无权限执行 {func.__name__},"
                    f"需要角色: {allowed_roles},当前角色: {user.get('role')}"
                )
            return func(user, *args, **kwargs)
        return wrapper
    return decorator


# 模拟用户数据
admin_user = {'name': '张三', 'role': 'admin'}
normal_user = {'name': '李四', 'role': 'user'}
guest_user = {'name': '王五', 'role': 'guest'}


@require_role('admin')
def delete_user(operator, target_user_id):
    """删除用户(仅管理员)"""
    return f"用户 {target_user_id} 已被 {operator['name']} 删除"


@require_role('admin', 'user')
def update_profile(user, user_id, new_data):
    """更新个人资料(管理员和普通用户)"""
    return f"用户 {user_id} 的资料已更新"


# 测试权限验证
try:
    print(delete_user(admin_user, 1001))
except PermissionDeniedError as e:
    print(f"错误: {e}")

try:
    print(delete_user(normal_user, 1001))
except PermissionDeniedError as e:
    print(f"错误: {e}")

try:
    print(update_profile(normal_user, 1002, {'age': 25}))
except PermissionDeniedError as e:
    print(f"错误: {e}")

六、装饰器性能监控应用

import time
import functools
from collections import defaultdict


class PerformanceMonitor:
    """性能监控器"""

    def __init__(self):
        self.call_counts = defaultdict(int)
        self.total_times = defaultdict(float)

    def track(self, func):
        """性能跟踪装饰器"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            end = time.perf_counter()

            self.call_counts[func.__name__] += 1
            self.total_times[func.__name__] += (end - start)

            return result
        return wrapper

    def report(self):
        """生成性能报告"""
        print("\n=== 性能统计报告 ===")
        for func_name in sorted(self.call_counts.keys()):
            count = self.call_counts[func_name]
            total_time = self.total_times[func_name]
            avg_time = total_time / count if count > 0 else 0
            print(
                f"{func_name:20s} | 调用次数: {count:4d} | "
                f"总耗时: {total_time:8.4f}s | 平均: {avg_time:8.4f}s"
            )


# 创建监控器实例
monitor = PerformanceMonitor()


@monitor.track
def heavy_computation(n):
    """模拟耗时计算"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total


@monitor.track
def quick_lookup(data, key):
    """模拟快速查询"""
    return data.get(key, None)


# 执行一些操作
for _ in range(5):
    heavy_computation(10000)

for _ in range(100):
    quick_lookup({'a': 1, 'b': 2}, 'a')

# 生成性能报告
monitor.report()

七、总结

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

1. 实现关注点分离:将日志、权限验证、性能监控等横切关注点从业务逻辑中分离出来
2. 提高代码复用性:一个装饰器可以应用到多个函数上
3. 保持代码整洁:避免在每个函数中重复相同的代码
4. 动态增强功能:在不修改原函数的情况下添加新功能

掌握装饰器不仅能让代码更加优雅,还能显著提升开发效率。建议在实际项目中多思考哪些场景适合使用装饰器,逐步将其融入到你的代码风格中。

记住:装饰器的核心思想是"函数也是对象,可以像变量一样传递和返回"。理解这一点,你就掌握了装饰器的精髓。

相关文章

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

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

Python 装饰器实战:从入门到精通

装饰器本质上是一个接受函数作为参数并返回新函数的高阶函数。理解这一点是掌握装饰器的关键。当你看到@decorator 语法时,Python 实际上是在执行 func = decorator(func)...

Python 上下文管理器实战指南:优雅处理资源的艺术

# Python 上下文管理器实战指南:优雅处理资源的艺术 在 Python 编程中,资源的获取与释放是一个永恒的主题。文件操作、数据库连接、网络请求、锁的获取...这些场景都遵循相同的模式:打开资...

Python 装饰器实战与原理深度解析

在 Python 开发中,我们经常需要在多个函数中添加相同的功能,比如日志记录、性能计时、权限校验等。如果每个函数都重复编写这些代码,不仅效率低下,还容易出错。装饰器正是为了解决这类问题而诞生的。...

Python 装饰器完全指南:从基础到实战应用的深度解析

一、装饰器的基本原理 装饰器的本质是一个接收函数作为参数,并返回一个新函数的高阶函数。Python 使用 @ 语法糖来应用装饰器,这让代码更加简洁易读。 让我们从一个最简单的计时装饰器开始,理解装...

Python 装饰器实战技巧:让代码更优雅高效

装饰器是 Python 中最优雅的特性之一,它允许我们在不修改函数源代码的情况下,动态地增强函数的功能。这种"切面编程"的思想让代码更加模块化、可维护。本文将带你深入理解装饰器的各种用法和实战技巧。一...

发表评论

访客

看不清,换一张

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