核心内容摘要
手把手复现CTFA框架：基于ViT的弱监督遥感分割从数据准备到结果可视化
在异步 HTTP 请求场景中aiohttp 是 Python 生态下的主流选择。
实际开发中请求的日志记录排查问题和失败重试提升稳定性是必备能力而 aiohttp 的中间件机制能优雅地实现这两个功能无需侵入业务代码。
本文将详细讲解如何基于 aiohttp 中间件打造通用的异步请求日志与重试组件。
核心概念aiohttp 中间件aiohttp 中间件是一个异步函数它介于ClientSession和实际 HTTP 请求之间能拦截请求的发起、响应的返回以及异常的抛出。
其核心作用是对请求 / 响应生命周期进行统一处理格式如下python运行async def middleware(app, handler): async def wrapper(request): # 请求发送前的处理如日志、参数修改 response await handler(request) # 执行实际请求 # 响应返回后的处理如日志、响应解析 return response return wrapper中间件的优势在于一次定义全局生效所有通过ClientSession发起的请求都会经过中间件处理。
实现思路分析我们需要实现两个核心功能且需保证逻辑解耦日志中间件记录请求的 URL、方法、状态码、耗时、请求体可选、响应体可选等关键信息重试中间件捕获指定类型的异常如网络超时、5xx 服务器错误按照配置的次数和间隔重试请求执行顺序重试中间件应包裹日志中间件先重试再记录最终的请求结果。
完整代码实现
1 依赖安装首先确保安装 aiohttp 和异步重试依赖tenacity支持异步重试比手动写循环更优雅bash运行pip install aiohttp tenacity python-dotenv
2 核心代码python运行import asyncio import logging import time from typing import Dict, Any, Optional import aiohttp from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, AsyncRetrying, ) # 配置日志格式 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler()] ) logger logging.getLogger(aiohttp-requests) # -------------------------- 日志中间件 -------------------------- async def logging_middleware(app: aiohttp.web.Application, handler): 记录请求的关键信息URL、方法、状态码、耗时、异常等 async def wrapper(request: aiohttp.ClientRequest): start_time time.time() request_id freq-{int(start_time *
} # 生成唯一请求ID try: # 执行实际请求 response await handler(request) # 计算耗时 elapsed round((time.time() - start_time) * 1000,
# 记录成功日志 logger.info( f[{request_id}] SUCCESS | {request.method} {request.url} | fStatus: {response.status} | Elapsed: {elapsed}ms ) return response except Exception as e: # 记录失败日志 elapsed round((time.time() - start_time) * 1000,
logger.error( f[{request_id}] FAILED | {request.method} {request.url} | fError: {str(e)} | Elapsed: {elapsed}ms, exc_infoTrue # 打印异常堆栈便于排查 ) raise # 抛出异常让重试中间件处理 return wrapper # -------------------------- 重试中间件 -------------------------- async def retry_middleware(app: aiohttp.web.Application, handler): 对指定异常的请求进行重试 - 重试次数3次 - 等待策略指数退避1s, 2s, 4s - 重试条件网络超时、5xx错误、连接错误 async def wrapper(request: aiohttp.ClientRequest): # 定义重试策略 retryer AsyncRetrying( stopstop_after_attempt(
, # 最多重试3次 waitwait_exponential(multiplier1, min1, max
, # 指数等待 retryretry_if_exception_type( ( aiohttp.ClientTimeoutError, # 超时异常 aiohttp.ClientConnectionError, # 连接异常 aiohttp.ServerConnectionError, # 服务器连接异常 ) ), before_sleeplambda retry_state: logger.warning( fRetry {retry_state.attempt_number} for {request.method} {request.url} fdue to {retry_state.outcome.exception()} ), # 重试前打印日志 ) try: # 执行带重试的请求 response await retryer.__aenter__() try: return await handler(request) finally: await retryer.__aexit__(None, None, None) except Exception as e: logger.error(fAll retries failed for {request.method} {request.url}: {str(e)}) raise return wrapper # -------------------------- 扩展ClientSession -------------------------- class CustomClientSession(aiohttp.ClientSession): 封装带日志和重试的ClientSession简化使用 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 注册中间件注意顺序重试在外日志在内 self._middlewares [retry_middleware, logging_middleware] # -------------------------- 测试代码 -------------------------- async def test_requests(): 测试异步请求日志与重试功能 # 使用自定义的Session async with CustomClientSession(timeoutaiohttp.ClientTimeout(total
) as session: # 测试正常请求 try: async with session.get(https://httpbin.org/get) as resp: print(fNormal request: {resp.status}) except Exception as e: print(fNormal request failed: {e}) # 测试会触发重试的请求模拟超时/5xx try: async with session.get(https://httpbin.org/delay/
as resp: # 超时请求 print(fRetry request: {resp.status}) except Exception as e: print(fRetry request failed: {e}) if __name__ __main__: asyncio.run(test_requests())
代码关键说明
1 日志中间件生成唯一请求 ID便于关联同一次请求的日志排查问题时可快速定位记录核心指标请求方法、URL、状态码、耗时失败时打印异常堆栈不吞异常记录日志后重新抛出异常保证重试中间件能捕获并处理。
2 重试中间件基于tenacity实现异步重试相比手动写循环tenacity支持更灵活的重试策略指数退避、最大次数、自定义条件精准重试仅对网络超时、连接错误等临时异常重试避免对业务异常如 400 参数错误无效重试重试日志每次重试前打印日志便于观察重试过程。
3 自定义 ClientSession封装中间件注册逻辑业务代码只需使用CustomClientSession无需重复注册中间件中间件顺序重试中间件在外日志中间件在内保证每次重试的请求都能被日志记录。
进阶优化
1 可配置化将重试次数、等待时间、日志级别等配置抽离到配置文件如.env便于不同环境调整python运行# 从环境变量读取配置 import os from dotenv import load_dotenv load_dotenv() RETRY_MAX_ATTEMPTS int(os.getenv(RETRY_MAX_ATTEMPTS,
) RETRY_WAIT_MULTIPLIER float(os.getenv(RETRY_WAIT_MULTIPLIER,
)
2 支持 5xx 状态码重试默认重试只捕获异常可扩展为对 5xx 响应码重试python运行# 在重试中间件中新增判断逻辑 async def wrapper(request: aiohttp.ClientRequest): async def _handle_request(): response await handler(request) if response.status 500: raise aiohttp.ServerErrorResponse( fServer error: {response.status}, statusresponse.status ) return response # 将原handler替换为_handle_request retryer AsyncRetrying(...) try: response await retryer.__aenter__() try: return await _handle_request() finally: await retryer.__aexit__(None, None, None) # ...
3 忽略指定 URL 重试对不需要重试的 URL如写操作接口添加白名单python运行# 重试中间件中添加过滤逻辑 NO_RETRY_URLS [/api/write, /api/submit] if any(url in str(request.url) for url in NO_RETRY_URLS): return await handler(request) # 直接执行不重试
使用
注意事项重试幂等性仅对幂等请求如 GET 查询重试POST/PUT 等写操作需确保接口幂等避免重复提交超时设置单个请求的超时时间不宜过长结合重试次数避免整体请求耗时过久日志脱敏若请求包含敏感信息如 token、密码需在日志中脱敏避免泄露中间件顺序重试中间件需在日志中间件外层否则重试的请求不会被日志记录。
总结aiohttp 中间件能无侵入地实现请求日志和重试核心是通过拦截handler执行过程添加通用逻辑日志中间件重点记录请求 ID、耗时、状态码和异常重试中间件基于tenacity实现精准的异步重试使用时需注意中间件顺序、重试幂等性和超时控制保证功能可靠且不引入新问题。
通过上述实现你可以快速为 aiohttp 异步请求添加企业级的日志和重试能力提升项目的可维护性和稳定性。