AIOHTTP 中间件请求日志

示例说明

请求级中间件是横切逻辑的标准挂载点：日志在处理器前后各占一半——前半取请求 ID 与起始时间，后半补状态与耗时。健康探针每几秒一次，进日志只会制造噪音，显式跳过。未捕获异常在中间件里兜底转 500 并记完整堆栈，请求 ID 回写响应头方便用户报障关联。

技术要点

• 装饰器 @web.middleware 声明中间件
• 探针路径免记防噪音
• 请求 ID 回写响应头闭环关联

代码解读

request_logging 先短路探针，再取上游请求 ID 或截断 UUID；异常分支区分 HTTPException（正常业务语义放行）与未知异常（记堆栈回 500）。成功路径用 perf_counter 计毫秒耗时，日志行包含方法、路径、状态、耗时与客户端地址。

使用建议

日志采样策略按流量级别调整，全量记录只在排障期临时开启。中间件顺序决定执行序，日志放最外层。慢请求阈值告警可在此埋点。

代码示例

import logging
import time
import uuid

from aiohttp import web

logger = logging.getLogger('request')
logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s %(name)s %(message)s',
)

routes = web.RouteTableDef()


@web.middleware
async def request_logging(request: web.Request, handler):
    if request.path == '/healthz':  # 探针流量不进日志
        return await handler(request)

    request_id = request.headers.get('X-Request-Id', str(uuid.uuid4())[:8])
    started = time.perf_counter()

    try:
        response = await handler(request)
    except web.HTTPException:
        raise
    except Exception:
        logger.exception(
            'request failed rid=%s path=%s', request_id, request.path
        )
        return web.json_response({'error': 'internal error'}, status=500)

    elapsed_ms = (time.perf_counter() - started) * 1000
    logger.info(
        'rid=%s %s %s -> %d %.1fms client=%s',
        request_id, request.method, request.path,
        response.status, elapsed_ms,
        request.remote or '-',
    )
    response.headers['X-Request-Id'] = request_id
    return response


@routes.get('/orders/{id}')
async def get_order(request: web.Request) -> web.Response:
    if request.match_info['id'] == '404':
        raise web.HTTPNotFound(text='no such order')
    return web.json_response({'id': request.match_info['id']})


app = web.Application(middlewares=[request_logging])
app.add_routes(routes)
web.run_app(app, port=8080)