怎么用Python装饰器封装日志记录与性能计时工具函数

一、装饰器拦截模型与核心原理

Python装饰器的本质是一个接收函数作为输入、并返回新包装函数的高阶函数。当我们在目标函数上方写上 @timer_and_logger 时，解释器在执行定义期会自动将该函数替换为装饰器返回的闭包函数。

在调用链路上，调用方发起的函数执行首先会进入装饰器内部定义的 wrapper 包装函数。包装函数在此处完成三个核心动作：记录调用发生的时间戳与参数详情、执行真实的业务函数逻辑、计算耗时并将结果与执行统计通过日志模块格式化输出。这种非侵入式设计遵循软件工程中的单一职责原则与开闭原则。

二、基础版计时日志装饰器的实现步骤

实现一个生产环境可用的基础装饰器，需要依序处理元数据保留、参数拦截、高精度计时和异常安全处理四个关键环节。

1. 为什么必须使用 functools.wraps

当被修饰的函数被 wrapper 替换后，其内置属性 __name__ 会变成 "wrapper"，文档注释 __doc__ 也会变成包装函数的注释。这会导致调试、日志追踪以及使用反射机制的框架（例如 FastAPI、Flask 路由绑定或 Sphinx 文档生成器）无法读取真实函数的元信息。functools.wraps 是标准库提供的辅助装饰器，它会自动将被修饰函数的元数据拷贝到外层包装函数上。

2. 为什么选用 time.perf_counter 而非 time.time

在性能计时场景下，不能使用 time.time()。因为 time.time() 返回的是系统时钟时间（Wall Clock Time），易受到操作系统网络授时校准（NTP同步）或人工修改系统时间的影响，可能产生时间倒流或跳变。而 time.perf_counter() 是基于CPU时钟的高精度单调时钟（Monotonic Clock），保证其参考点始终单调向前递增，是统计程序代码块运行耗时的官方推荐函数。

3. 基础代码实现

import logging
import time
from functools import wraps

# 配置基础日志格式
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger("perf_tool")

def log_and_time(func):
    """
    基础日志与耗时统计装饰器
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 1. 获取并格式化参数预览
        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)
        
        # 截断过长参数，防止巨型对象挤爆日志
        display_sig = signature if len(signature) < 100 else signature[:97] + "..."
        
        logger.info(f"▶ 开始调用 [{func.__name__}] 参数: ({display_sig})")
        
        # 2. 记录单调时钟起点
        start_time = time.perf_counter()
        try:
            # 3. 执行原函数
            result = func(*args, **kwargs)
            return result
        except Exception as e:
            # 异常时记录现场并重新抛出，保障业务感知
            cost_ms = (time.perf_counter() - start_time) * 1000
            logger.error(f"✖ 调用异常 [{func.__name__}] 耗时: {cost_ms:.2f}ms, 异常类型: {type(e).__name__}, 原因: {e}")
            raise
        finally:
            # 4. 无论成功失败，计算最终耗时（若未进except）
            cost_ms = (time.perf_counter() - start_time) * 1000
            # 仅在无未捕获异常时输出正常结束日志
            if 'result' in locals():
                logger.info(f"✔ 调用完成 [{func.__name__}] 耗时: {cost_ms:.2f}ms")

    return wrapper

三、进阶封装：支持阈值报警与自定义参数

在实际生产环境中，所有函数执行都打印日志可能会导致日志量暴增。常见需求是：只对执行耗时超过一定阈值（如 200ms）的“慢函数”输出 WARNING 日志，或者允许不同的业务模块传入各自定制的 Logger 实例。

带参数的装饰器比普通装饰器多一层闭包：最外层是“装饰器工厂函数”，负责接收配置参数；中间层是真正的装饰器函数，接收被修饰的 func；最内层则是拦截调用的 wrapper。

带参配置装饰器完整代码

import logging
import time
from functools import wraps

def monitor(slow_threshold_ms=200.0, logger_instance=None):
    """
    通用性能与日志监控装饰器工厂
    :param slow_threshold_ms: 慢调用阈值（毫秒），超过该值触发警告日志
    :param logger_instance: 自定义日志记录器，未提供则使用根记录器
    """
    def decorator(func):
        # 确定使用的 logger
        active_logger = logger_instance or logging.getLogger(func.__module__)

        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                cost_ms = (time.perf_counter() - start) * 1000
                
                # 根据耗时情况输出不同级别的日志
                if cost_ms >= slow_threshold_ms:
                    active_logger.warning(
                        f"⚡ [慢函数警报] {func.__qualname__} 耗时: {cost_ms:.2f}ms (设定阈值: {slow_threshold_ms}ms)"
                    )
                else:
                    active_logger.info(
                        f"✔ [正常执行] {func.__qualname__} 耗时: {cost_ms:.2f}ms"
                    )
                return result
            except Exception as err:
                cost_ms = (time.perf_counter() - start) * 1000
                active_logger.error(
                    f"✖ [执行崩溃] {func.__qualname__} 耗时: {cost_ms:.2f}ms, 错误: {err}",
                    exc_info=True
                )
                raise
        return wrapper
    return decorator

四、实际应用示例与结果验证

通过编写一个模拟业务脚本，测试正常调用、模拟慢查询以及异常触发三种典型场景，观察装饰器的实际输出表现。

# 业务示例代码
import time

# 场景 1：普通轻量任务（设定慢函数阈值为100ms）
@monitor(slow_threshold_ms=100.0)
def fetch_user_profile(user_id: int):
    time.sleep(0.02)  # 耗时约 20ms
    return {"id": user_id, "name": "Alice"}

# 场景 2：耗时较大的复杂统计计算
@monitor(slow_threshold_ms=100.0)
def calculate_monthly_report():
    time.sleep(0.15)  # 耗时约 150ms，会超过阈值
    return {"total_sales": 88000}

# 场景 3：触发意外错误的方法
@monitor(slow_threshold_ms=100.0)
def query_remote_service():
    time.sleep(0.01)
    raise ConnectionResetError("远端API连接重置")

if __name__ == "__main__":
    fetch_user_profile(1001)
    calculate_monthly_report()
    try:
        query_remote_service()
    except ConnectionResetError:
        pass  # 外层捕获处理

执行上述脚本后，终端或日志收集系统将清晰呈现如下信息（示例输出）：

2026-10-24 23:50:01 [INFO] ✔ [正常执行] fetch_user_profile 耗时: 20.45ms
2026-10-24 23:50:01 [WARNING] ⚡ [慢函数警报] calculate_monthly_report 耗时: 151.12ms (设定阈值: 100.0ms)
2026-10-24 23:50:01 [ERROR] ✖ [执行崩溃] query_remote_service 耗时: 10.32ms, 错误: 远端API连接重置
Traceback (most recent call last):
  File "demo.py", line 28, in query_remote_service
    raise ConnectionResetError("远端API连接重置")
ConnectionResetError: 远端API连接重置

五、工程实践中的注意事项与避坑指南

装饰器在带来便利的同时，若处理不当，也会引入意想不到的系统风险。以下为四个在工程落地中必须注意的关键点：

• 入参与返回值序列化开销：如果被修饰函数的入参包含大型二进制流、几万条记录的Pandas DataFrame或大型字典，盲目执行 str(args) 会消耗大量CPU和内存，甚至使日志记录本身比业务逻辑还要慢。在记录参数时，应做长度限制或仅记录核心ID字段。
• 不要吞没异常：装饰器应当如实记录异常，并通过 raise 向上抛出。除非业务明确要求“降级容错”，否则装饰器绝不能用空 except 吞掉异常并返回 None，否则调用方将无法感知下游失败。
• 类方法中的 self/cls 绑定：当装饰器用于类的方法（Method）时，第一个参数为 self 或 cls。使用 *args, **kwargs 可以自然兼容实例方法与类方法。利用 func.__qualname__ 而非 func.__name__，能够在日志中打印出 UserService.fetch_user_profile 包含类名的完整路径，提升定位效率。
• 异步协程函数（async def）的兼容性：普通的 def wrapper 无法直接装饰 async def 异步函数。如果直接修饰异步函数，result = func(*args, **kwargs) 返回的仅仅是一个未执行的协程对象（Coroutine），导致计时仅测量了创建协程对象的几微秒，且无法真正等待执行完成。对于异步函数，需要编写异步包装器。

六、方案总结与选型参考

在实际项目中，可以根据系统的复杂度与性能诉求，选择适合的装饰器封装深度：

通过合理运用 functools.wraps、time.perf_counter() 与多层闭包结构，我们可以将原本分散在各处的日志打点与性能排查逻辑收敛为干净、可复用的模块工具。在提升代码可读性与健壮性的同时，为系统运维和线上性能分析提供精确、可靠的指标支撑。