Python装饰器(Decorators)是Python语言中一个强大而优雅的特性,它允许我们在不修改原有函数代码的情况下,动态地改变函数的行为。装饰器本质上是一个高阶函数,它接收一个函数作为参数并返回一个新的函数。这种设计模式在Python开发中被广泛应用,特别是在日志记录、性能测试、权限验证、缓存等场景中。本文将从装饰器的基本概念开始,逐步深入到装饰器的各种高级用法和实际应用场景,帮助你全面掌握这一重要技术。

装饰器的基本概念和语法

装饰器的核心思想是函数作为一等公民(First-class citizens),这意味着函数可以像其他对象一样被传递、赋值和返回。装饰器使用Python的@符号语法糖,使代码更加简洁易读。

最简单的装饰器示例

让我们从一个最基础的装饰器开始:

def my_decorator(func):
    def wrapper():
        print("函数执行前")
        func()
        print("函数执行后")
    return wrapper

@my_decorator
def say_hello():
    print("Hello!")

# 调用函数
say_hello()

输出结果:

函数执行前
Hello!
函数执行后

代码解析:

  1. my_decorator 是一个装饰器函数,它接收一个函数 func 作为参数
  2. 在内部定义了一个 wrapper 函数,它会在调用时执行额外的逻辑
  3. wrapper 函数中调用了原始函数 func()
  4. @my_decorator 语法糖等价于执行 say_hello = my_decorator(say_hello)
  5. 当调用 say_hello() 时,实际上执行的是 wrapper() 函数

带参数的函数装饰器

上面的例子只能装饰无参数的函数。如果要装饰带参数的函数,需要使用 *args**kwargs

def greet_decorator(func):
    def wrapper(*args, **kwargs):
        print("准备问候...")
        result = func(*args, **kwargs)
        print("问候完成!")
        return result
    return wrapper

@greet_decorator
def greet(name, greeting="Hello"):
    print(f"{greeting}, {name}!")

greet("Alice", greeting="Hi")

输出结果:

准备问候...
Hi, Alice!
问候完成!

装饰器的常见应用场景

1. 日志记录(Logging)

装饰器非常适合用于记录函数的调用信息:

import logging
import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def log_execution_time(func):
    """记录函数执行时间和日志"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        logging.info(f"开始执行函数: {func.__name__}")
        
        try:
            result = func(*args, **kwargs)
            end_time = time.time()
            execution_time = end_time - start_time
            logging.info(f"函数 {func.__name__} 执行成功,耗时: {execution_time:.4f}秒")
            return result
        except Exception as e:
            logging.error(f"函数 {func.__name__} 执行失败: {str(e)}")
            raise
    return wrapper

@log_execution_time
def calculate_sum(n):
    """计算1到n的和"""
    total = sum(range(1, n+1))
    time.sleep(0.5)  # 模拟耗时操作
    return total

# 测试
result = calculate_sum(1000000)
print(f"计算结果: {result}")

2. 权限验证(Authentication)

在Web开发中,装饰器常用于检查用户权限:

class User:
    def __init__(self, username, role):
        self.username = username
        self.role = role

def require_admin(func):
    """需要管理员权限的装饰器"""
    def wrapper(user, *args, **kwargs):
        if user.role != "admin":
            raise PermissionError(f"用户 {user.username} 需要管理员权限才能执行此操作")
        return func(user, *args, **kwargs)
    return wrapper

def require_login(func):
    """需要登录的装饰器"""
    def wrapper(user, *args, **kwargs):
        if user is None:
            raise PermissionError("用户未登录")
        return func(user, *args, **kwargs)
    return wrapper

@require_login
@require_admin
def delete_user(user, target_username):
    """删除用户(仅管理员可用)"""
    print(f"用户 {user.username} 成功删除了用户 {target_username}")

# 测试
admin_user = User("admin", "admin")
normal_user = User("bob", "user")

try:
    delete_user(admin_user, "someone")  # 成功
    delete_user(normal_user, "someone")  # 失败
except PermissionError as e:
    print(f"错误: {e}")

3. 缓存(Memoization)

装饰器可以用于缓存函数的计算结果,避免重复计算:

import time

def cache(seconds):
    """带过期时间的缓存装饰器"""
    def decorator(func):
        cache_dict = {}
        def wrapper(*args):
            current_time = time.time()
            # 检查缓存是否存在且未过期
            if args in cache_dict:
                result, timestamp = cache_dict[args]
                if current_time - timestamp < seconds:
                    print(f"使用缓存结果: {args}")
                    return result
            
            # 执行函数并缓存结果
            print(f"计算新结果: {args}")
            result = func(*args)
            cache_dict[args] = (result, current_time)
            return result
        return wrapper
    return decorator

@cache(5)  # 缓存5秒
def fibonacci(n):
    """计算斐波那契数列"""
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 测试
print(fibonacci(10))
time.sleep(3)
print(fibonacci(10))  # 使用缓存
time.sleep(6)
print(fibonacci(10))  # 重新计算

装饰器的高级用法

1. 带参数的装饰器

有时候我们需要给装饰器本身传递参数,这需要三层嵌套函数:

def repeat(num_times):
    """重复执行函数指定次数的装饰器"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            results = []
            for i in range(num_times):
                print(f"第 {i+1} 次执行:")
                result = func(*args, **kwargs)
                results.append(result)
            return results
        return wrapper
    return decorator

@repeat(3)
def greet(name):
    print(f"Hello, {name}!")
    return name

# 测试
results = greet("World")
print(f"返回结果: {results}")

2. 类装饰器

除了函数装饰器,Python还支持类装饰器。类装饰器需要实现 __call__ 方法:

class CountCalls:
    """统计函数调用次数的类装饰器"""
    def __init__(self, func):
        self.func = func
        self.num_calls = 0
    
    def __call__(self, *args, **kwargs):
        self.num_calls += 1
        print(f"函数 {self.func.__name__} 被调用了 {self.num_calls} 次")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello!")

# 测试
say_hello()
say_hello()
say_hello()

3. 装饰器的叠加顺序

当多个装饰器叠加时,它们的执行顺序是从下往上的:

def decorator1(func):
    def wrapper(*args, **kwargs):
        print("装饰器1 - 前")
        result = func(*args, **kwargs)
        print("装饰器1 - 后")
        return result
    return wrapper

def decorator2(func):
    def wrapper(*args, **kwargs):
        print("装饰器2 - 前")
        result = func(*args, **kwargs)
        print("装饰器2 - 后")
        return result
    return wrapper

@decorator1
@decorator2
def my_function():
    print("执行核心函数")

my_function()

输出结果:

装饰器1 - 前
装饰器2 - 前
执行核心函数
装饰器2 - 后
装饰器1 - 后

4. 保留原函数元信息

默认情况下,装饰器会替换原函数的 __name____doc__ 等元信息。使用 functools.wraps 可以解决这个问题:

import functools

def my_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        """包装函数的文档字符串"""
        print("装饰器逻辑")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def example():
    """原始函数的文档字符串"""
    pass

print(example.__name__)  # 输出: example
print(example.__doc__)   # 输出: 原始函数的文档字符串

实际项目中的装饰器应用

1. API限流装饰器

import time
from collections import defaultdict

class RateLimiter:
    """API限流装饰器"""
    def __init__(self, max_requests, period):
        self.max_requests = max_requests
        self.period = period
        self.calls = defaultdict(list)
    
    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取调用者标识(这里简单使用IP,实际中可能更复杂)
            caller = kwargs.get('ip', 'default')
            now = time.time()
            
            # 清理过期的调用记录
            self.calls[caller] = [t for t in self.calls[caller] if now - t < self.period]
            
            # 检查是否超过限制
            if len(self.calls[caller]) >= self.max_requests:
                raise Exception(f"API调用频率超过限制,请等待{self.period}秒")
            
            # 记录本次调用
            self.calls[caller].append(now)
            return func(*args, **kwargs)
        return wrapper

# 使用示例
limiter = RateLimiter(max_requests=3, period=10)

@limiter
def api_call(ip):
    print(f"API调用成功,IP: {ip}")
    return {"status": "success"}

# 测试
for i in range(5):
    try:
        api_call(ip="192.168.1.1")
    except Exception as e:
        print(f"第{i+1}次调用失败: {e}")

2. 事务管理装饰器

class Transaction:
    """数据库事务装饰器"""
    def __init__(self, db_connection):
        self.db = db_connection
    
    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                # 开始事务
                self.db.begin_transaction()
                print("事务开始")
                
                # 执行业务逻辑
                result = func(*args, **kwargs)
                
                # 提交事务
                self.db.commit()
                print("事务提交成功")
                return result
                
            except Exception as e:
                # 回滚事务
                self.db.rollback()
                print(f"事务回滚: {e}")
                raise
        return wrapper

# 模拟数据库连接
class MockDB:
    def begin_transaction(self):
        pass
    
    def commit(self):
        pass
    
    def rollback(self):
        pass

db = MockDB()

@Transaction(db)
def transfer_money(from_account, to_account, amount):
    # 模拟转账操作
    print(f"从账户 {from_account} 转账 {amount} 到账户 {to_account}")
    if amount < 0:
        raise ValueError("转账金额不能为负")
    return True

# 测试
transfer_money("A123", "B456", 100)

装饰器的最佳实践

1. 保持装饰器的单一职责

每个装饰器应该只做一件事,避免在一个装饰器中实现过多功能。

2. 使用functools.wraps

始终使用 @functools.wraps 来保留原函数的元信息,便于调试和文档生成。

3. 考虑装饰器的性能影响

装饰器会增加函数调用的开销,在性能敏感的场景中需要权衡。

4. 提供清晰的错误信息

当装饰器的条件不满足时,应该提供明确的错误信息。

5. 编写装饰器的单元测试

装饰器本身也需要测试,确保其逻辑正确。

import unittest

class TestDecorator(unittest.TestCase):
    def test_log_execution_time(self):
        @log_execution_time
        def test_func():
            return 42
        
        # 测试函数是否正常工作
        result = test_func()
        self.assertEqual(result, 42)

if __name__ == '__main__':
    unittest.main()

总结

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

  • 增强函数功能:在不修改原函数代码的情况下添加新功能
  • 提高代码复用性:将通用逻辑封装成装饰器,可以在多个函数中重复使用
  • 保持代码整洁:通过分离关注点,使业务逻辑更加清晰
  • 支持函数式编程:体现了函数作为一等公民的设计理念

掌握装饰器的使用,不仅能让你写出更优雅、更Pythonic的代码,还能帮助你更好地理解Python的高级特性。从简单的日志记录到复杂的权限控制,装饰器在实际项目中有着广泛的应用场景。希望本文能帮助你全面理解并熟练运用Python装饰器。