• 装饰器（Decorators）是一种强大而灵活的功能，用于修改或增强函数或类的行为

• 装饰器本质上是一个函数，它接受另一个函数或类作为参数，并返回一个新的函数或类

• 它们通常用于在不修改原始代码的情况下添加额外的功能

• 装饰器的语法使用@符号，将装饰器应用于目标函数或类

下面是一些常见且好用的装饰器：

1、缓存装饰器

import functools


@functools.lru_cache(maxsize=None)
def example_function(x, y):
    """
    Python自带的缓存装饰器
    """
    return x + y


def memory_cache(expire_time: int = 60):
    """
    内存缓存装饰器

    :param expire_time: 缓存过期时间(秒),默认60秒
    :return: 装饰器函数
    """
    cache = {}

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            key = str(args) + str(kwargs)
            if key in cache:
                result, timestamp = cache[key]
                if time.time() - timestamp < expire_time:
                    return result
            result = func(*args, **kwargs)
            cache[key] = (result, time.time())
            return result

        return wrapper

    return decorator


def redis_cache(expire_time: int = 60):
    """
    Redis缓存装饰器

    :param expire_time: 缓存过期时间(秒),默认60秒
    :return: 装饰器函数
    """

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            result = redis_client.get(key)
            if result:
                return result
            result = func(*args, **kwargs)
            redis_client.setex(key, expire_time, result)
            return result

        return wrapper

    return decorator

2、日志装饰器

def enhanced_logger(
    show_args: bool = True,  # 是否记录函数参数
    show_return: bool = True,  # 是否记录返回值
    time_threshold: float = 0.3,  # 执行时间阈值，超过此阈值将以警告级别记录
    context_key: str = None,  # 上下文信息的键名，如果提供，将尝试从第一个参数中获取此属性作为上下文
) -> Callable:
    """增强版函数调用日志

    :param show_args: 是否记录函数参数
    :param show_return: 是否记录返回值
    :param time_threshold: 执行时间阈值，超过此阈值将以警告级别记录
    :param context_key: 上下文信息的键名，如果提供，将尝试从第一个参数中获取此属性作为上下文
    :return: 装饰器
    """

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # 获取上下文信息
            context = None
            if context_key and args:
                context = getattr(args[0], context_key, None)

            extra = {context_key: context} if context else {}

            # 参数记录
            if show_args:
                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)
                logger.debug(f"调用函数 {func.__name__}({signature})", extra=extra)
            else:
                logger.debug(f"调用函数 {func.__name__}", extra=extra)

            # 记录开始时间
            start_time = time.perf_counter()

            try:
                # 执行函数
                result = func(*args, **kwargs)

                # 计算执行时间
                elapsed = time.perf_counter() - start_time

                # 根据阈值记录执行时间
                if elapsed > time_threshold:
                    logger.warning(f"{func.__name__} 执行耗时 {elapsed:.4f}秒 (阈值 {time_threshold}秒)", extra=extra)
                else:
                    logger.info(f"{func.__name__} 执行耗时 {elapsed:.4f}秒", extra=extra)

                # 返回值记录
                if show_return:
                    logger.debug(f"{func.__name__} 返回值类型: {type(result).__name__}, 值: {result!r}", extra=extra)
                else:
                    logger.debug(f"{func.__name__} 执行完成", extra=extra)

                return result

            except Exception as e:
                # 计算执行时间
                elapsed = time.perf_counter() - start_time

                # 记录异常信息，保留原始堆栈
                logger.exception(f"函数 {func.__name__} 执行异常，耗时 {elapsed:.4f}秒: {str(e)}", extra=extra)

                # 重新抛出异常，保留原始堆栈
                raise

        return wrapper

    return decorator

3、重试装饰器

def retry(max_attempts, delay):
    """
    重试装饰器
    :param max_attempts: 最大重试次数（retries）
    :param delay: 重试间隔（秒）
    :return: 装饰器函数
    """

    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"{func.__name__} 第 {attempts + 1} 次执行失败，将在 {delay} 秒后重试...")
                    attempts += 1
                    time.sleep(delay)
            raise Exception("超过最大重试次数，请求失败！")

        return wrapper

    return decorator

4、限流装饰器

def rate_limit(max_calls: int, period: int):
    """
    限流装饰器

    Args:
        max_calls: 最大调用次数
        period: 限制周期（秒）
    """

    def decorator(func: Callable) -> Callable:
        last_called = time.time() - period

        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            nonlocal last_called
            if time.time() - last_called < period / max_calls:
                time.sleep((period / max_calls) - (time.time() - last_called))
            result = func(*args, **kwargs)
            last_called = time.time()
            return result

        return wrapper

    return decorator

5、超时装饰器

def timeout(seconds: int):
    """
    超时装饰器
    Args:
        seconds: 超时时间（秒）
    """

    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            result = None
            try:
                result = func(*args, **kwargs)
            except Exception as e:  # 捕获所有异常
                logger.error(f"Function {func.__name__} timed out after {seconds} seconds")
            return result

        return wrapper

    return decorator

6、耗时装饰器

def timer(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        logger.info(f"{func.__name__} 耗时: {end_time - start_time:.2f} 秒")
        return result

    return wrapper

7、进度装饰器

import time
from tqdm import trange


def bar(desc="", unit="it"):
    def decorator(func):
        def inner(*args, **kwargs):
            pbar = None
            gen = func(*args, **kwargs)

            try:
                while True:
                    i = next(gen)
                    if pbar is None:
                        pbar = trange(i, desc=desc, unit=unit)
                    pbar.update(1)
            except StopIteration as e:
                pbar.close()
                return e.value

        return inner

    return decorator

8、类型检查装饰器

def type_check(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 获取函数的参数注解
        annotations = func.__annotations__

        # 遍历参数和注解，检查类型是否正确
        for arg, annotation in zip(args, annotations.values()):
            if not isinstance(arg, annotation):
                raise TypeError(f"参数 {arg} 的类型应为 {annotation}，但实际类型为 {type(arg)}")

        # 调用原始函数
        return func(*args, **kwargs)

    return wrapper

9、返回结果装饰器

def send_to_url(url: str, max_retries: int = 3, retry_interval: int = 2, timeout: int = 10):
    """
    装饰器：将函数返回的结果发送到指定URL

    @param url: 目标服务器URL
    @param max_retries: 最大重试次数
    @param retry_interval: 重试间隔（秒）
    @param timeout: 请求超时时间（秒）
    :return: 装饰器函数
    """

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # 执行原始函数获取结果
            result = func(*args, **kwargs)

            # 发送结果到指定URL
            send_success = _send_result(result, url, max_retries, retry_interval, timeout)

            # 返回原始结果和发送状态
            return result, send_success

        return wrapper

    return decorator


def _send_result(result: Any, url: str, max_retries: int = 3, retry_interval: int = 2, timeout: int = 10) -> bool:
    """
    使用POST请求，将结果发送到指定服务器，自动重连

    @param result: 结果
    @param url: 目标服务器URL
    @param max_retries: 最大重试次数
    @param retry_interval: 重试间隔（秒）
    @param timeout: 请求超时时间（秒）
    :return: 发送状态：True/False
    """
    # 尝试将结果转换为JSON
    try:
        if hasattr(result, "__dict__"):
            # 如果结果是对象，尝试转换其属性为字典
            payload = result.__dict__
        elif isinstance(result, dict):
            # 如果已经是字典，直接使用
            payload = result
        else:
            # 尝试使用json序列化
            payload = json.loads(json.dumps(result))
    except (TypeError, json.JSONDecodeError) as e:
        logger.error(f"结果序列化失败: {str(e)}")
        return False

    # 添加时间戳
    payload["timestamp"] = time.time()

    # 设置请求头
    headers = {"Content-Type": "application/json", "Accept": "application/json"}

    # 重试逻辑
    for attempt in range(1, max_retries + 1):
        try:
            logger.info(f"正在发送结果到 {url}，第 {attempt} 次尝试...")
            response = requests.post(url=url, json=payload, headers=headers, timeout=timeout)

            # 检查响应状态码
            if response.status_code in (200, 201, 202):
                try:
                    response_data = response.json()
                    logger.info(f"发送成功，服务器响应: {response_data}")
                    return True
                except json.JSONDecodeError:
                    logger.error(f"发送成功，但服务器响应不是有效的JSON: {response.text}")
                    return True
            else:
                logger.info(f"请求失败，状态码: {response.status_code}, 响应: {response.text}")

                # 如果是最后一次尝试，返回失败
                if attempt == max_retries:
                    return False

                # 否则等待后重试
                time.sleep(retry_interval)

        except RequestException as e:
            logger.warning(f"请求异常: {str(e)}")

            # 如果是最后一次尝试，返回失败
            if attempt == max_retries:
                return False

            # 否则等待后重试
            logger.warning(f"将在 {retry_interval} 秒后重试...")
            time.sleep(retry_interval)

    # 如果所有尝试都失败
    return False