装饰器是Python中最具魅力的特性之一，它用简洁的语法实现了代码的横向扩展。本文通过10个真实开发场景，带你从入门到精通这个"魔法工具"。每个案例都包含问题背景、解决方案和源码解析，让你轻松掌握装饰器的核心思想。

一、日志记录：给函数加上"黑匣子"
在系统运维中，我们经常需要记录函数的调用情况。传统方式是在每个函数里写日志代码，而装饰器可以优雅地解决这个问题。

import time
import logging

def log_duration(func):
logging.basicConfig(level=logging.INFO)

def wrapper(*args, **kwargs):
    start_time = time.time()
    result = func(*args, **kwargs)
    duration = time.time() - start_time

    logging.info(f"{func.__name__} executed in {duration:.4f}s")
    return result
return wrapper

@log_duration
def process_data(data):
time.sleep(1) # 模拟耗时操作
return [x*2 for x in data]

process_data(range(1000))

输出: INFO:root:process_data executed in 1.0023s

实现原理：装饰器在函数执行前后插入计时逻辑，通过args和*kwargs保持原函数参数不变。这种非侵入式设计让日志功能与业务逻辑完全解耦。

进阶技巧：添加日志级别参数

def log_duration(level=logging.INFO):
def decorator(func):
def wrapper(args, **kwargs):
start = time.time()
result = func(args, **kwargs)
logging.log(level, f"{func.name} took {time.time()-start:.2f}s")
return result
return wrapper
return decorator

@log_duration(logging.DEBUG)
def heavy_computation():
time.sleep(2)

二、权限验证：API网关的守门人
在Web开发中，装饰器是实现权限控制的利器。Flask/Django等框架都大量使用装饰器进行路由保护。

from functools import wraps

def require_auth(role):
def decorator(func):
@wraps(func) # 保留原函数元信息
def wrapper(user, args, **kwargs):
if user.get('role') != role:
raise PermissionError(f"Requires {role} role")
return func(user, args, **kwargs)
return wrapper
return decorator

class User:
def init(self, role):
self.role = role

@require_auth('admin')
def delete_user(user, user_id):
print(f"Deleting user {user_id}")

admin = User('admin')
guest = User('guest')

delete_user(admin, 123) # 正常执行
delete_user(guest, 123) # 抛出PermissionError

关键点：

functools.wraps保留原函数名和文档字符串
嵌套装饰器实现参数化（require_auth接收role参数）
清晰的错误提示帮助快速定位问题
三、性能测试：找出代码中的"蜗牛"
当系统变慢时，我们需要快速定位性能瓶颈。这个装饰器可以自动统计函数调用次数和平均耗时。

from collections import defaultdict
import time

class PerformanceMonitor:
def init(self):
self.stats = defaultdict(lambda: {'count': 0, 'total': 0})

def __call__(self, func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        duration = time.time() - start

        self.stats[func.__name__]['count'] += 1
        self.stats[func.__name__]['total'] += duration

        return result
    return wrapper

def report(self):
    print("\nPerformance Report:")
    for name, data in self.stats.items():
        avg = data['total'] / data['count']
        print(f"{name}: {data['count']} calls, avg {avg:.4f}s")

monitor = PerformanceMonitor()

@monitor
def slow_function():
time.sleep(0.5)

@monitor
def fast_function():
time.sleep(0.1)

for _ in range(3):
slow_function()
fast_function()

monitor.report()

输出示例：

Performance Report:

slow_function: 3 calls, avg 0.5003s

fast_function: 3 calls, avg 0.1001s

设计思路：

使用类装饰器保存统计状态
defaultdict简化计数逻辑
分离统计和报告功能，符合单一职责原则
四、缓存机制：给函数装上"记忆大脑"
对于计算密集型函数，缓存结果可以大幅提升性能。这个装饰器实现了简单的LRU缓存策略。

from functools import wraps

def memoize(maxsize=128):
cache = {}

def decorator(func):
    @wraps(func)
    def wrapper(*args):
        if args not in cache:
            if len(cache) >= maxsize:
                # 简单实现：移除第一个元素（实际应使用LRU算法）
                removed_key = next(iter(cache))
                del cache[removed_key]
            cache[args] = func(*args)
        return cache[args]
    return wrapper
return decorator

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

print([fibonacci(x) for x in range(10)])

实际只计算了fibonacci(0)-fibonacci(9)各一次

优化方向：

使用functools.lru_cache替代（Python内置实现）
添加TTL（过期时间）支持
处理可变参数的哈希问题
五、重试机制：让网络请求更健壮
处理不稳定API时，自动重试可以显著提高成功率。这个装饰器实现了指数退避重试策略。

import time
import random
from functools import wraps

def retry(max_attempts=3, delay=1, backoff=2):
def decorator(func):
@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
wait_time = delay * (backoff (attempts-1))
time.sleep(wait_time + random.uniform(0, 0.1*wait_time))
return wrapper
return decorator

@retry(max_attempts=5, delay=0.5)
def unreliable_api_call():
if random.random() < 0.7: # 70%概率失败
raise ConnectionError("API unavailable")
return "Success"

print(unreliable_api_call()) # 最终成功概率约99.76%

数学原理：

经过n次重试后，成功概率 = 1 - (失败概率)^(n+1)
示例中5次重试后成功概率 = 1 - 0.7^6 ≈ 99.76%
六、参数校验：给函数加上"安全带"
在数据处理管道中，参数校验是重要防线。这个装饰器可以验证输入参数类型和范围。

from functools import wraps

def validate_params(validators):
def decorator(func):
@wraps(func)
def wrapper(*args, kwargs):

        # 将位置参数转为关键字参数（简化示例）
        bound_args = func.__code__.co_varnames[:func.__code__.co_argcount]
        kw = dict(zip(bound_args, args))
        kw.update(kwargs)

        for param, validator in validators.items():
            if param not in kw:
                continue
            value = kw[param]
            if not validator(value):
                raise ValueError(f"Invalid value {value} for parameter {param}")

        return func(*args, **kwargs)
    return wrapper
return decorator

def is_positive(x):
return isinstance(x, (int, float)) and x > 0

@validate_params(age=is_positive, name=lambda x: len(x)>0)
def create_user(name, age):
print(f"Creating user: {name}, {age} years old")

create_user("Alice", 30) # 正常执行
create_user("", -5) # 抛出ValueError

扩展建议：

使用inspect模块获取更准确的参数信息
添加更丰富的验证器（如正则表达式、枚举值等）
集成Pydantic等验证库
七、异步支持：让装饰器跟上异步潮流
在异步编程中，装饰器需要特殊处理才能正确工作。这个例子展示了如何编写异步装饰器。

import asyncio
from functools import wraps

def async_retry(max_attempts=3):
def decorator(func):
@wraps(func)
async def wrapper(args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return await func(args, kwargs)
except Exception as e:
last_exception = e
if attempt == max_attempts - 1:
raise
await asyncio.sleep(0.1 * (2 attempt)) # 指数退避
raise last_exception
return wrapper
return decorator

@async_retry(max_attempts=5)
async def fetch_data():
if random.random() < 0.8: # 80%概率失败
raise ConnectionError("Network error")
return {"data": "sample"}

async def main():
result = await fetch_data()
print(result)

asyncio.run(main())

关键区别：

使用async def定义包装器
使用await调用被装饰函数
异步等待重试延迟
八、单例模式：确保类只有一个实例
装饰器也可以用于实现设计模式。这个单例装饰器比传统实现更简洁。

def singleton(cls):
instances = {}

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):
print("Initializing database connection...")
self.connection_id = id(self)

conn1 = DatabaseConnection()
conn2 = DatabaseConnection()

print(conn1 is conn2) # 输出: True
print(conn1.connection_id == conn2.connection_id) # 输出: True

线程安全改进：

import threading

def thread_safe_singleton(cls):
instances = {}
lock = threading.Lock()

def get_instance(*args, **kwargs):
    nonlocal instances
    if cls not in instances:
        with lock:
            if cls not in instances:  # 双重检查
                instances[cls] = cls(*args, **kwargs)
    return instances[cls]

return get_instance

九、上下文管理：自动资源处理
这个装饰器自动处理文件打开/关闭，避免资源泄漏。

from functools import wraps

def open_file(mode='r', encoding='utf-8'):
def decorator(func):
@wraps(func)
def wrapper(file_path, args, **kwargs):
with open(file_path, mode=mode, encoding=encoding) as f:
return func(f, args, **kwargs)
return wrapper
return decorator

@open_file('r')
def count_lines(filehandle):
return sum(1 for in file_handle)

line_count = count_lines("sample.txt")

扩展应用：

def db_transaction(func):
@wraps(func)
def wrapper(args, *kwargs):
try:

        # 假设begin_transaction/commit/rollback是已实现的函数
        begin_transaction()
        result = func(*args, **kwargs)
        commit()
        return result
    except:
        rollback()
        raise
return wrapper

十、插件系统：动态扩展应用功能
这个装饰器实现了简单的插件注册机制，让应用功能可以动态扩展。

class PluginSystem:
def init(self):
self.plugins = []

def plugin(self, func):
    self.plugins.append(func)
    return func  # 保持装饰器可链式调用

def execute_plugins(self, *args, **kwargs):
    results = []
    for plugin in self.plugins:
        try:
            results.append(plugin(*args, **kwargs))
        except Exception as e:
            print(f"Plugin {plugin.__name__} failed: {e}")
    return results

创建插件系统实例

system = PluginSystem()

注册插件

@system.plugin
def multiply(x, y):
return x * y

@system.plugin
def add(x, y):
return x + y

@system.plugin
def faulty_plugin(x, y):
return 10 / 0 # 会触发异常处理

执行所有插件

results = system.execute_plugins(3, 4)
print(results) # 输出: [12, 7, None] (最后一个插件失败返回None)

装饰器最佳实践总结

保持简洁：单个装饰器不超过20行代码
明确命名：使用@retry而非@with_retry
合理嵌套：避免超过3层的装饰器嵌套
文档完整：每个装饰器都应包含docstring说明
性能考虑：避免在装饰器中做耗时操作
异常处理：明确装饰器是否应该捕获异常
类型提示：Python 3.6+推荐添加类型注解
from typing import Callable, Any, TypeVar, Optional

T = TypeVar('T')

def robust_decorator(func: Callable[..., T]) -> Callable[..., T]:
"""增强函数健壮性的装饰器示例"""
def wrapper(args, **kwargs) -> T:
try:
return func(args, **kwargs)
except Exception as e:
print(f"Function {func.name} failed: {str(e)}")
raise # 根据实际需求决定是否重新抛出
return wrapper

装饰器是Python中"小而美"的典范，它用简洁的语法解决了大量横切关注点问题。通过合理运用装饰器，我们可以写出更模块化、更易维护的代码。希望本文的10个实战案例能帮助你掌握这个强大工具，在实际项目中发挥它的威力。

​