深入理解Python装饰器链与AOP思想实现

面向切面编程（Aspect-Oriented Programming，AOP）是一种编程范式，旨在将横切关注点（如日志、权限、事务）从业务逻辑中分离。Python装饰器是实现AOP思想的强大工具。

装饰器基础概念

装饰器本质上是一个高阶函数，它接收一个函数作为参数，并返回一个新的函数。通过装饰器，我们可以在不修改原函数代码的情况下，为其添加额外功能。

AOP核心实现

"""
Python装饰器链与AOP思想实现
包含日志、权限、缓存、重试、限流等切面功能
"""

import functools
import time
import logging
from typing import Callable, Any, List, Dict, Optional
from functools import wraps
import hashlib
import json
from dataclasses import dataclass
from enum import Enum

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


# ============ 基础装饰器 ============
def timing_decorator(func: Callable) -> Callable:
    """计时装饰器 - 记录函数执行时间"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start_time
        logger.info(f"[Timing] {func.__name__} 执行时间: {elapsed:.4f}s")
        return result
    return wrapper


def logging_decorator(func: Callable) -> Callable:
    """日志装饰器 - 记录函数调用信息"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"[Log] 调用 {func.__name__}, args={args}, kwargs={kwargs}")
        result = func(*args, **kwargs)
        logger.info(f"[Log] {func.__name__} 返回: {result}")
        return result
    return wrapper


def retry_decorator(max_attempts: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
    """重试装饰器 - 失败时自动重试"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts:
                        logger.error(f"[Retry] {func.__name__} 重试{max_attempts}次后仍失败: {e}")
                        raise
                    logger.warning(f"[Retry] {func.__name__} 第{attempt}次尝试失败，{delay}s后重试...")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator


# ============ 带参数的装饰器 ============
def cache_decorator(ttl: int = 300, maxsize: int = 128):
    """缓存装饰器 - 带过期时间的缓存"""
    cache = {}
    cache_times = {}
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            key_parts = [func.__name__] + list(args) + [f"{k}={v}" for k, v in sorted(kwargs.items())]
            key = hashlib.md5(json.dumps(key_parts, sort_keys=True, default=str).encode()).hexdigest()
            
            now = time.time()
            
            # 检查缓存是否有效
            if key in cache:
                if now - cache_times[key] < ttl:
                    logger.info(f"[Cache] {func.__name__} 命中缓存")
                    return cache[key]
                else:
                    # 缓存过期，删除
                    del cache[key]
                    del cache_times[key]
            
            # 执行函数并缓存结果
            result = func(*args, **kwargs)
            
            # 清理旧缓存
            if len(cache) >= maxsize:
                oldest_key = min(cache_times, key=cache_times.get)
                del cache[oldest_key]
                del cache_times[oldest_key]
            
            cache[key] = result
            cache_times[key] = now
            logger.info(f"[Cache] {func.__name__} 缓存结果")
            
            return result
        
        # 添加缓存管理方法
        wrapper.cache_clear = lambda: cache.clear() or cache_times.clear()
        wrapper.cache_info = lambda: {"size": len(cache), "maxsize": maxsize, "ttl": ttl}
        
        return wrapper
    return decorator


def rate_limit_decorator(max_calls: int = 10, period: float = 60.0):
    """限流装饰器 - 控制函数调用频率"""
    def decorator(func: Callable) -> Callable:
        calls = []
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # 清理过期记录
            calls[:] = [call_time for call_time in calls if now - call_time < period]
            
            if len(calls) >= max_calls:
                wait_time = period - (now - calls[0])
                raise Exception(f"[RateLimit] 调用过于频繁，请等待{wait_time:.1f}秒")
            
            calls.append(now)
            return func(*args, **kwargs)
        
        return wrapper
    return decorator


# ============ 权限控制装饰器 ============
class Permission(Enum):
    """权限枚举"""
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"


@dataclass
class User:
    """用户类"""
    id: int
    username: str
    permissions: List[Permission]


def require_permission(*required_permissions: Permission):
    """权限检查装饰器"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 获取当前用户（实际项目中从上下文获取）
            current_user = kwargs.get('current_user') or (args[0] if args else None)
            
            if not current_user:
                raise PermissionError("[Auth] 未登录用户")
            
            user_perms = set(current_user.permissions)
            required = set(required_permissions)
            
            if not required.issubset(user_perms):
                missing = required - user_perms
                raise PermissionError(f"[Auth] 缺少权限: {[p.value for p in missing]}")
            
            logger.info(f"[Auth] 用户 {current_user.username} 权限检查通过")
            return func(*args, **kwargs)
        return wrapper
    return decorator


# ============ 事务装饰器 ============
def transaction_decorator(func: Callable) -> Callable:
    """事务装饰器 - 模拟数据库事务"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"[Transaction] 开始事务: {func.__name__}")
        try:
            result = func(*args, **kwargs)
            logger.info(f"[Transaction] 提交事务: {func.__name__}")
            return result
        except Exception as e:
            logger.error(f"[Transaction] 回滚事务: {func.__name__}, 错误: {e}")
            raise
    return wrapper


# ============ 装饰器链组合 ============
def compose_decorators(*decorators):
    """组合多个装饰器"""
    def decorator(func):
        for dec in reversed(decorators):
            func = dec(func)
        return func
    return decorator


# ============ AOP切面基类 ============
class Aspect:
    """切面基类"""
    
    def before(self, func_name: str, args: tuple, kwargs: dict):
        """前置通知"""
        pass
    
    def after(self, func_name: str, args: tuple, kwargs: dict, result: Any):
        """后置通知"""
        pass
    
    def around(self, func: Callable, args: tuple, kwargs: dict):
        """环绕通知"""
        self.before(func.__name__, args, kwargs)
        result = func(*args, **kwargs)
        self.after(func.__name__, args, kwargs, result)
        return result
    
    def on_exception(self, func_name: str, args: tuple, kwargs: dict, exception: Exception):
        """异常通知"""
        pass


def apply_aspect(aspect: Aspect):
    """应用切面"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return aspect.around(func, args, kwargs)
            except Exception as e:
                aspect.on_exception(func.__name__, args, kwargs, e)
                raise
        return wrapper
    return decorator


# ============ 具体切面实现 ============
class LoggingAspect(Aspect):
    """日志切面"""
    
    def before(self, func_name: str, args: tuple, kwargs: dict):
        logger.info(f"[Aspect-Before] {func_name} 开始执行")
    
    def after(self, func_name: str, args: tuple, kwargs: dict, result: Any):
        logger.info(f"[Aspect-After] {func_name} 执行完成，结果: {result}")
    
    def on_exception(self, func_name: str, args: tuple, kwargs: dict, exception: Exception):
        logger.error(f"[Aspect-Exception] {func_name} 发生异常: {exception}")


class PerformanceAspect(Aspect):
    """性能监控切面"""
    
    def __init__(self):
        self.timings = {}
    
    def before(self, func_name: str, args: tuple, kwargs: dict):
        self.timings[func_name] = time.time()
    
    def after(self, func_name: str, args: tuple, kwargs: dict, result: Any):
        if func_name in self.timings:
            elapsed = time.time() - self.timings[func_name]
            logger.info(f"[Aspect-Performance] {func_name} 耗时: {elapsed:.4f}s")


# ============ 业务服务类 ============
class OrderService:
    """订单服务 - 展示装饰器应用"""
    
    @logging_decorator
    @timing_decorator
    @cache_decorator(ttl=60)
    def get_order(self, order_id: int) -> Dict:
        """获取订单信息"""
        time.sleep(0.1)  # 模拟数据库查询
        return {"id": order_id, "status": "completed", "amount": 199.99}
    
    @transaction_decorator
    @logging_decorator
    @require_permission(Permission.WRITE)
    def create_order(self, current_user: User, product_id: int, quantity: int) -> Dict:
        """创建订单"""
        time.sleep(0.2)
        return {"order_id": 12345, "product_id": product_id, "quantity": quantity}
    
    @retry_decorator(max_attempts=3, delay=0.5)
    @rate_limit_decorator(max_calls=5, period=60)
    def process_payment(self, order_id: int) -> bool:
        """处理支付"""
        import random
        if random.random() < 0.3:  # 30%概率失败
            raise Exception("支付服务暂时不可用")
        return True
    
    @apply_aspect(LoggingAspect())
    @apply_aspect(PerformanceAspect())
    def cancel_order(self, order_id: int) -> bool:
        """取消订单 - 使用AOP切面"""
        time.sleep(0.05)
        return True


class UserService:
    """用户服务"""
    
    @cache_decorator(ttl=300)
    def get_user_profile(self, user_id: int) -> Dict:
        """获取用户资料"""
        time.sleep(0.1)
        return {"id": user_id, "name": f"User_{user_id}", "email": f"user{user_id}@example.com"}
    
    @require_permission(Permission.ADMIN)
    def delete_user(self, current_user: User, user_id: int) -> bool:
        """删除用户 - 需要管理员权限"""
        logger.info(f"[Admin] 用户 {current_user.username} 删除用户 {user_id}")
        return True


# ============ 演示主函数 ============
def main():
    """主函数 - 演示装饰器和AOP"""
    print("="*60)
    print("Python装饰器链与AOP思想实现")
    print("="*60)
    
    # 创建服务实例
    order_service = OrderService()
    user_service = UserService()
    
    # 创建测试用户
    admin_user = User(id=1, username="admin", permissions=[Permission.READ, Permission.WRITE, Permission.ADMIN])
    normal_user = User(id=2, username="alice", permissions=[Permission.READ, Permission.WRITE])
    guest_user = User(id=3, username="guest", permissions=[Permission.READ])
    
    # 1. 缓存装饰器演示
    print("\n【缓存装饰器演示】")
    order1 = order_service.get_order(1001)
    order2 = order_service.get_order(1001)  # 命中缓存
    print(f"缓存信息: {order_service.get_order.cache_info()}")
    
    # 2. 权限控制演示
    print("\n【权限控制演示】")
    try:
        order_service.create_order(admin_user, product_id=101, quantity=2)
        order_service.create_order(guest_user, product_id=102, quantity=1)  # 应该失败
    except PermissionError as e:
        print(f"权限错误: {e}")
    
    # 3. 重试和限流演示
    print("\n【重试和限流演示】")
    try:
        for i in range(7):
            try:
                result = order_service.process_payment(1001)
                print(f"支付处理结果: {result}")
            except Exception as e:
                print(f"支付处理失败: {e}")
    except Exception as e:
        print(f"限流触发: {e}")
    
    # 4. AOP切面演示
    print("\n【AOP切面演示】")
    order_service.cancel_order(1001)
    
    # 5. 管理员权限演示
    print("\n【管理员权限演示】")
    try:
        user_service.delete_user(admin_user, 999)
        user_service.delete_user(normal_user, 999)  # 应该失败
    except PermissionError as e:
        print(f"权限错误: {e}")
    
    print("\n" + "="*60)
    print("装饰器链与AOP总结")
    print("="*60)
    print("1. 装饰器实现了横切关注点的分离")
    print("2. 可以组合多个装饰器形成装饰器链")
    print("3. AOP切面提供了更灵活的拦截机制")
    print("4. 常用切面: 日志、权限、缓存、事务、重试、限流")
    print("5. 保持代码DRY原则，提高可维护性")
    print("="*60)


if __name__ == "__main__":
    main()

装饰器执行流程图

AOP切面模型

关键要点

1. 装饰器本质：高阶函数，接收函数返回函数
2. functools.wraps：保留原函数的元信息
3. 装饰器链：多个装饰器从内到外依次应用
4. AOP优势：横切关注点与业务逻辑分离
5. 常用场景：日志、缓存、权限、事务、重试、限流

通过合理使用装饰器和AOP思想，可以编写出更加模块化、可维护的代码。