Orion K Blog
Python装饰器实战：15个实用案例详解
Orion K Lv6
  2023-02-18 16:45:00  

Python装饰器实战：15个实用案例详解

Python装饰器是我最喜欢的语言特性之一，它让我们能够优雅地修改或增强函数和类的行为，而无需修改其源代码。在这篇文章中，我将分享15个实用的装饰器案例，帮助你更好地理解和应用这一强大的功能。

装饰器基础

在深入案例之前，让我们快速回顾一下装饰器的基本概念：

1
2
3
4
5
6
7
8
9
10
11
def my_decorator(func):
    def wrapper(*args, **kwargs):
        print("在函数调用前执行")
        result = func(*args, **kwargs)
        print("在函数调用后执行")
        return result
    return wrapper

@my_decorator
def say_hello(name):
    print(f"Hello, {name}")

现在，让我们看看一些实用的装饰器案例。

1. 计时装饰器

测量函数执行时间：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import time
import functools

def timer(func):
    @functools.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:.6f}秒")
        return result
    return wrapper

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

2. 重试装饰器

自动重试可能失败的操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import time
import functools
import random

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts == max_attempts:
                        raise
                    print(f"尝试 {attempts}/{max_attempts} 失败: {e}. 等待 {delay} 秒后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=5, delay=2)
def unstable_network_call():
    if random.random() < 0.7:  # 70% 失败率
        raise ConnectionError("网络连接失败")
    return "成功获取数据"

3. 缓存装饰器

缓存函数结果以提高性能：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import functools

def memoize(func):
    cache = {}
    
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 将kwargs转换为可哈希的形式
        key_kwargs = tuple(sorted(kwargs.items()))
        key = (args, key_kwargs)
        
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    
    return wrapper

@memoize
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

4. 参数验证装饰器

验证函数参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import functools

def validate_types(**expected_types):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取函数参数名
            func_code = func.__code__
            param_names = func_code.co_varnames[:func_code.co_argcount]
            
            # 合并位置参数和关键字参数
            all_args = dict(zip(param_names, args))
            all_args.update(kwargs)
            
            # 验证类型
            for param, expected_type in expected_types.items():
                if param in all_args:
                    actual_value = all_args[param]
                    if not isinstance(actual_value, expected_type):
                        raise TypeError(f"参数 '{param}' 必须是 {expected_type.__name__} 类型, "
                                       f"但收到的是 {type(actual_value).__name__}")
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

@validate_types(name=str, age=int)
def greet(name, age):
    return f"你好, {name}! 你已经 {age} 岁了。"

5. 日志装饰器

记录函数调用和结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import functools
import logging

logging.basicConfig(level=logging.INFO)

def log_function(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        args_repr = [repr(a) for a in args]
        kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
        signature = ", ".join(args_repr + kwargs_repr)
        
        logging.info(f"调用 {func.__name__}({signature})")
        try:
            result = func(*args, **kwargs)
            logging.info(f"{func.__name__} 返回: {result!r}")
            return result
        except Exception as e:
            logging.exception(f"{func.__name__} 抛出异常: {e}")
            raise
    return wrapper

@log_function
def divide(a, b):
    return a / b

6. 单例装饰器

确保类只有一个实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def singleton(cls):
    instances = {}
    
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    
    return get_instance

@singleton
class DatabaseConnection:
    def __init__(self, host, username, password):
        self.host = host
        self.username = username
        self.password = password
        print(f"连接到数据库 {host}")

7. 权限检查装饰器

检查用户权限：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import functools

def require_permission(permission):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(user, *args, **kwargs):
            if permission not in user.permissions:
                raise PermissionError(f"用户 {user.username} 没有 {permission} 权限")
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

class User:
    def __init__(self, username, permissions=None):
        self.username = username
        self.permissions = permissions or []

@require_permission("admin")
def delete_user(current_user, user_id):
    print(f"{current_user.username} 删除了用户 {user_id}")

8. 限速装饰器

限制函数调用频率：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import time
import functools
from collections import deque

def rate_limit(max_calls, period):
    """限制函数在指定时间段内的最大调用次数"""
    calls = deque(maxlen=max_calls)
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # 移除过期的调用记录
            while calls and calls[0] < now - period:
                calls.popleft()
                
            # 检查是否超过调用限制
            if len(calls) >= max_calls:
                wait_time = calls[0] + period - now
                raise Exception(f"调用频率过高，请在 {wait_time:.2f} 秒后重试")
                
            result = func(*args, **kwargs)
            calls.append(now)
            return result
        return wrapper
    return decorator

@rate_limit(max_calls=3, period=60)
def send_api_request(endpoint):
    print(f"发送请求到 {endpoint}")
    # 实际API调用代码

9. 超时装饰器

限制函数执行时间：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import signal
import functools

class TimeoutError(Exception):
    pass

def timeout(seconds):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            def handler(signum, frame):
                raise TimeoutError(f"函数执行超过 {seconds} 秒")
            
            # 设置信号处理器
            original_handler = signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            
            try:
                result = func(*args, **kwargs)
            finally:
                # 恢复原始信号处理器
                signal.alarm(0)
                signal.signal(signal.SIGALRM, original_handler)
                
            return result
        return wrapper
    return decorator

@timeout(5)
def long_running_task():
    import time
    time.sleep(10)  # 这将触发超时

10. 异步重试装饰器

异步函数的重试机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import asyncio
import functools

def async_retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts == max_attempts:
                        raise
                    print(f"尝试 {attempts}/{max_attempts} 失败: {e}. 等待 {delay} 秒后重试...")
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@async_retry(max_attempts=3, delay=2)
async def fetch_data(url):
    # 模拟网络请求
    if random.random() < 0.7:  # 70% 失败率
        raise ConnectionError("网络连接失败")
    return f"来自 {url} 的数据"

11. 参数转换装饰器

自动转换函数参数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import functools

def convert_args(**converters):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取函数参数名
            func_code = func.__code__
            param_names = func_code.co_varnames[:func_code.co_argcount]
            
            # 转换位置参数
            converted_args = list(args)
            for i, param_name in enumerate(param_names[:len(args)]):
                if param_name in converters:
                    converted_args[i] = converters[param_name](args[i])
            
            # 转换关键字参数
            converted_kwargs = {}
            for key, value in kwargs.items():
                if key in converters:
                    converted_kwargs[key] = converters[key](value)
                else:
                    converted_kwargs[key] = value
            
            return func(*converted_args, **converted_kwargs)
        return wrapper
    return decorator

@convert_args(age=int, height=float)
def register_user(name, age, height):
    print(f"注册用户: {name}, 年龄: {age} ({type(age)}), 身高: {height} ({type(height)})")

# 即使传入字符串，也会自动转换
register_user("Alice", "30", "165.5")

12. 属性缓存装饰器

缓存类属性计算结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def cached_property(func):
    attr_name = f"_{func.__name__}"
    
    @property
    @functools.wraps(func)
    def wrapper(self):
        if not hasattr(self, attr_name):
            setattr(self, attr_name, func(self))
        return getattr(self, attr_name)
    
    return wrapper

class Circle:
    def __init__(self, radius):
        self.radius = radius
    
    @cached_property
    def area(self):
        print("计算面积...")
        import math
        return math.pi * self.radius ** 2

13. 弃用警告装饰器

标记弃用的函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import functools
import warnings

def deprecated(reason):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            warnings.warn(
                f"{func.__name__} 已弃用: {reason}",
                category=DeprecationWarning,
                stacklevel=2
            )
            return func(*args, **kwargs)
        return wrapper
    return decorator

@deprecated("请使用 new_function() 代替")
def old_function():
    return "旧功能"

14. 事件触发装饰器

在函数执行前后触发事件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import functools

class EventSystem:
    def __init__(self):
        self.listeners = {}
    
    def subscribe(self, event, callback):
        if event not in self.listeners:
            self.listeners[event] = []
        self.listeners[event].append(callback)
    
    def emit(self, event, *args, **kwargs):
        if event in self.listeners:
            for callback in self.listeners[event]:
                callback(*args, **kwargs)

event_system = EventSystem()

def emit_events(before_event=None, after_event=None):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if before_event:
                event_system.emit(before_event, *args, **kwargs)
            
            result = func(*args, **kwargs)
            
            if after_event:
                event_system.emit(after_event, result, *args, **kwargs)
            
            return result
        return wrapper
    return decorator

@emit_events(before_event="user_login_attempt", after_event="user_login_complete")
def login_user(username, password):
    # 验证逻辑
    return {"username": username, "success": True}

15. 参数注入装饰器

自动注入依赖：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import functools
import inspect

class DependencyContainer:
    def __init__(self):
        self.dependencies = {}
    
    def register(self, name, instance):
        self.dependencies[name] = instance
    
    def get(self, name):
        return self.dependencies.get(name)

container = DependencyContainer()

def inject(**dependencies):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取函数签名
            sig = inspect.signature(func)
            
            # 注入依赖
            for param_name, dependency_name in dependencies.items():
                if param_name in sig.parameters and param_name not in kwargs:
                    kwargs[param_name] = container.get(dependency_name)
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 注册依赖
container.register("db", {"connection": "sqlite:///:memory:"})
container.register("logger", {"log": lambda msg: print(f"LOG: {msg}")})

@inject(db="db", logger="logger")
def save_user(user, db, logger):
    logger["log"](f"保存用户到数据库: {user}")
    # 使用db进行实际操作
    return True

结论

装饰器是Python中非常强大的特性，可以帮助我们编写更简洁、更可维护的代码。通过这15个实用案例，你应该能够更好地理解装饰器的工作原理和应用场景。

记住，好的装饰器应该遵循单一职责原则，专注于解决一个特定的问题。此外，使用functools.wraps来保留原始函数的元数据也是一个好习惯。

你有什么创新的装饰器用例吗？欢迎在评论区分享！

  1. Python装饰器实战：15个实用案例详解
    1. 装饰器基础
    2. 1. 计时装饰器
    3. 2. 重试装饰器
    4. 3. 缓存装饰器
    5. 4. 参数验证装饰器
    6. 5. 日志装饰器
    7. 6. 单例装饰器
    8. 7. 权限检查装饰器
    9. 8. 限速装饰器
    10. 9. 超时装饰器
    11. 10. 异步重试装饰器
    12. 11. 参数转换装饰器
    13. 12. 属性缓存装饰器
    14. 13. 弃用警告装饰器
    15. 14. 事件触发装饰器
    16. 15. 参数注入装饰器
    17. 结论
© 2025    Orion K
本站由 提供部署服务
  1. Python装饰器实战：15个实用案例详解
    1. 装饰器基础
    2. 1. 计时装饰器
    3. 2. 重试装饰器
    4. 3. 缓存装饰器
    5. 4. 参数验证装饰器
    6. 5. 日志装饰器
    7. 6. 单例装饰器
    8. 7. 权限检查装饰器
    9. 8. 限速装饰器
    10. 9. 超时装饰器
    11. 10. 异步重试装饰器
    12. 11. 参数转换装饰器
    13. 12. 属性缓存装饰器
    14. 13. 弃用警告装饰器
    15. 14. 事件触发装饰器
    16. 15. 参数注入装饰器
    17. 结论