FastAPI集成aiohttp时如何校验请求头有效性并自动重发过期请求

问题2解答：aiohttp全局请求拦截实现方式 #

aiohttp本身没有内置的"中间件"定义，但有两种成熟方案可以实现全局请求逻辑注入：

• 官方推荐使用TraceConfig追踪配置，可以在请求发起前、响应接收后等全生命周期节点绑定自定义回调函数，实现全局拦截
• 更简单易用的方式是封装HttpClient类的请求方法，统一包装所有请求逻辑，侵入性低、调试成本小，适合当前的使用场景

问题1解答：认证头自动校验+过期重发实现 #

直接改造现有HttpClient类即可，无需修改上层业务逻辑的调用习惯，实现逻辑如下：

1. 内部存储当前认证信息、过期时间，提前预留缓冲时间避免临界时间失效
2. 封装统一的认证刷新方法，加锁避免并发场景下重复调用认证接口
3. 封装统一请求入口，自动注入有效认证头，遇到401错误时刷新认证后自动重发一次

改造后的HttpClient实现代码 #

import aiohttp
import time
import asyncio
from typing import Optional, Dict

class HttpClient:
    session: aiohttp.ClientSession = None
    # 认证信息存储
    _auth_key: Optional[str] = None
    _auth_sign: Optional[str] = None
    _auth_expire_at: float = 0
    # 并发刷新锁，避免多请求同时触发认证接口调用
    _refresh_lock: asyncio.Lock = asyncio.Lock()
    # 第三方认证配置
    AUTH_ENDPOINT = "https://third-party-service.com/api/auth"
    AUTH_EXPIRE_BUFFER: int = 10  # 提前10秒刷新认证，规避临界失效问题

    def start(self):
        self.session = aiohttp.ClientSession()

    async def stop(self):
        await self.session.close()
        self.session = None
        # 销毁时清空认证信息
        self._auth_key = None
        self._auth_sign = None
        self._auth_expire_at = 0

    def __call__(self) -> aiohttp.ClientSession:
        assert self.session is not None
        return self.session

    async def _refresh_auth(self) -> None:
        """请求第三方接口刷新认证信息，加锁避免并发重复调用"""
        async with self._refresh_lock:
            # 拿到锁之后再校验一次，避免等待锁的过程中已经被其他请求刷新过
            if time.time() < self._auth_expire_at:
                return
            # 此处替换为实际的第三方认证请求参数
            async with self.session.post(
                self.AUTH_ENDPOINT,
                json={"app_id": "your_app_id", "app_secret": "your_app_secret"}
            ) as resp:
                resp.raise_for_status()
                auth_data = await resp.json()
                self._auth_key = auth_data["auth_key"]
                self._auth_sign = auth_data["auth_sign"]
                # 根据接口返回的有效期计算过期时间，减去缓冲时间
                self._auth_expire_at = time.time() + auth_data["expires_in"] - self.AUTH_EXPIRE_BUFFER

    async def _build_auth_headers(self, custom_headers: Optional[Dict] = None) -> Dict:
        """构造带有效认证信息的请求头"""
        if time.time() >= self._auth_expire_at:
            await self._refresh_auth()
        headers = custom_headers.copy() if custom_headers else {}
        # 替换为实际的认证头字段名
        headers["X-Third-Auth-Key"] = self._auth_key
        headers["X-Third-Auth-Sign"] = self._auth_sign
        return headers

    async def request(self, method: str, url: str, headers: Optional[Dict] = None, **kwargs) -> aiohttp.ClientResponse:
        """统一请求入口，自动处理认证和重发逻辑"""
        # 第一次请求
        auth_headers = await self._build_auth_headers(headers)
        response = await self.session.request(method, url, headers=auth_headers, **kwargs)
        # 遇到401认证失效，刷新后重发一次
        if response.status == 401:
            response.close()
            await self._refresh_auth()
            auth_headers = await self._build_auth_headers(headers)
            response = await self.session.request(method, url, headers=auth_headers, **kwargs)
        return response

    # 快捷请求方法，适配原有调用习惯
    async def get(self, url: str, headers: Optional[Dict] = None, **kwargs) -> aiohttp.ClientResponse:
        return await self.request("GET", url, headers=headers, **kwargs)

    async def post(self, url: str, headers: Optional[Dict] = None, **kwargs) -> aiohttp.ClientResponse:
        return await self.request("POST", url, headers=headers, **kwargs)

    async def put(self, url: str, headers: Optional[Dict] = None, **kwargs) -> aiohttp.ClientResponse:
        return await self.request("PUT", url, headers=headers, **kwargs)

    async def delete(self, url: str, headers: Optional[Dict] = None, **kwargs) -> aiohttp.ClientResponse:
        return await self.request("DELETE", url, headers=headers, **kwargs)

http_client = HttpClient()

业务代码调用示例 #

上层业务代码几乎无需修改，直接调用封装好的请求方法即可：

from app.core.http_client import http_client
from fastapi import APIRouter

router = APIRouter()

@router.get("/some-endpoint")
async def some_endpoint():
    url = "http://third-party-service.com/api/data"
    custom_headers = {"Content-Type": "application/json"}
    # 无需手动处理认证头，直接调用封装的get方法即可
    async with await http_client.get(url, headers=custom_headers, params={"page": 1}) as response:
        response.raise_for_status()
        return await response.json()

补充说明 #

• 可以根据业务需求调整重发次数，当前实现最多重发1次，避免无限循环
• 可以在_refresh_auth方法中添加异常捕获逻辑，处理第三方认证接口调用失败的场景
• 如果需要自定义其他全局请求逻辑，直接修改request方法即可

内容的提问来源于stack exchange，提问作者kng