
本文介绍如何利用 Flask 的 before_request 和 after_request 钩子,结合应用上下文对象 g,自动计算每个请求的处理耗时,并将响应时间(如 0.45 s)结构化地注入日志行中,无需修改每个路由逻辑。
本文介绍如何利用 flask 的 `before_request` 和 `after_request` 钩子,结合应用上下文对象 `g`,自动计算每个请求的处理耗时,并将响应时间(如 `0.45 s`)结构化地注入日志行中,无需修改每个路由逻辑。
要在 Flask 中实现响应时间的自动测量与日志集成,核心思路是:在请求开始时记录时间戳,在响应返回前计算耗时,并将其注入日志格式或自定义日志输出中。由于原日志由 Werkzeug 默认服务器生成(如 [%(asctime)s][%(name)s][%(levelname)s][%(filename)s] ...),直接修改其日志格式无法动态插入响应时间(因其在 WSGI 层生成,早于 Flask 路由执行)。因此,推荐采用 Flask 应用层拦截 + 自定义结构化日志 的方式,既保持原有日志体系,又精准补充性能指标。
✅ 推荐方案:使用 before_request / after_request + 自定义日志格式
以下为完整可运行示例,兼容您已有的 setup_logger 配置,并确保响应时间精确到毫秒、格式统一(如 [0.452 s]):
from flask import Flask, g, request, current_app
import time
import logging
# 复用您的 logger 初始化(稍作增强)
def setup_logger(logp=None, debug=False):
log_level = logging.DEBUG if debug else logging.INFO
# 注意:此处扩展 format,预留 %(response_time)s 占位符
form = "[%(asctime)s][%(name)s][%(levelname)s][%(filename)s][%(response_time)s] %(message)s"
datefmt = "%Y-%m-%d-%H:%M:%S"
logging.basicConfig(level=log_level, format=form, datefmt=datefmt)
if logp is not None:
fhandler = logging.StreamHandler(open(logp, 'a'))
fhandler.setFormatter(logging.Formatter(form, datefmt))
logging.root.addHandler(fhandler)
logger = logging.getLogger("MY_APP")
setup_logger()
logger.debug(f"Import {__file__}")
app = Flask(__name__)
@app.before_request
def record_start_time():
"""在请求进入时记录起始时间"""
g.start_time = time.time()
@app.after_request
def log_response_time(response):
"""在响应发出前计算耗时并注入日志上下文"""
if hasattr(g, 'start_time'):
elapsed = time.time() - g.start_time
# 格式化为带单位的字符串,保留3位小数(如 "0.452 s")
response_time_str = f"{elapsed:.3f} s"
# 为当前请求临时注入日志上下文(需配合 LoggerAdapter 或 custom filter)
# 更简洁可靠的做法:使用 LoggerAdapter 动态注入字段
adapter = logging.LoggerAdapter(
logger,
extra={'response_time': f'[{response_time_str}]'}
)
# 记录含响应时间的访问日志(替代默认 werkzeug 日志,更可控)
adapter.info(
f'{request.remote_addr} - - [{request.date_string}] '
f'"{request.method} {request.path} {request.environ.get("SERVER_PROTOCOL", "HTTP/1.1")}" '
f'{response.status_code} -'
)
return response
# 示例路由(验证效果)
@app.route('/available/space', methods=['GET'])
def available_space():
time.sleep(0.45) # 模拟业务耗时
return {"status": "ok", "space": 1024}
@app.route('/health', methods=['GET'])
def health_check():
return {"status": "healthy"}
? 关键说明与注意事项
- 避免干扰 Werkzeug 日志:Werkzeug 的默认访问日志([werkzeug][INFO][_internal.py] ...)独立于 Flask 生命周期,无法直接注入动态字段。因此本方案主动接管访问日志输出,用 LoggerAdapter 确保 %(response_time)s 可被格式化器识别,同时保留时间、模块、级别等原始信息。
- 精度保障:使用 time.time()(而非 time.perf_counter())已足够满足秒级日志场景;若需微秒级分析,可改用 perf_counter() 并调整格式。
- 线程安全:g 对象是 Flask 请求上下文内的线程局部存储(Thread-Local),天然保证多请求并发下的隔离性,无需额外加锁。
-
异常兜底:若请求中途抛出异常(未走到 after_request),g.start_time 将残留。建议搭配 teardown_request 清理:
@app.teardown_request def cleanup_g(exception): g.pop('start_time', None) - 生产环境建议:对于高吞吐场景,可将耗时日志设为 DEBUG 级别,或通过配置开关控制是否启用,避免 I/O 成为瓶颈。
? 总结
通过 before_request 记录起点、after_request 计算并记录耗时,再借助 LoggerAdapter 动态注入日志字段,即可在不侵入业务代码的前提下,为所有 API 端点自动添加标准化的响应时间标记。最终日志形如:
[2024-04-19-04:20:02][MY_APP][INFO][app.py][0.452 s] 172.23.0.2 - - [Fri, 19 Apr 2024 04:20:02 GMT] "GET /available/space HTTP/1.1" 200 -
——清晰、可筛选、可监控,为性能分析与慢接口定位提供坚实基础。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











