FastAPI路由方法/端点之间传递数据的推荐实现方案
最优方案分场景选择,都可以规避你提到的额外组件开销问题
首先先明确完整请求链路的前提:第三方服务调用endpoint_1拿到唯一标识后,需要将该标识同步给前端,前端携带该标识调用endpoint_2才能完成数据匹配,这是跨独立请求传递数据的必要链路。
场景1:单实例部署、无分布式扩容需求
直接用内存级TTL缓存即可,不需要额外部署任何服务,实现成本最低。
你可以用纯Python实现的cachetools库的TTLCache,自带过期自动清理、容量上限控制,不会出现内存泄漏问题:
- 先安装依赖:
pip install cachetools
- 改造后的代码示例:
from urllib.parse import urlencode from uuid import uuid4 import httpx from fastapi import APIRouter, HTTPException from cachetools import TTLCache router = APIRouter() # 初始化缓存:最大存1000条记录,每条过期时间5分钟(可根据业务调整) attributes_cache = TTLCache(maxsize=1000, ttl=300) @router.post("/api/endpoint_1") async def endpoint_1(attributes: dict): # 生成唯一请求ID作为关联标识 request_id = str(uuid4()) attributes_cache[request_id] = attributes # 将request_id返回给第三方调用方 return {"request_id": request_id} @router.get("/api/endpoint_2") async def endpoint_2(request_id: str): # 校验请求ID是否存在 if request_id not in attributes_cache: raise HTTPException(status_code=404, detail="请求ID不存在或已过期") # 取出后直接删除缓存,避免占用内存 attributes = attributes_cache.pop(request_id) url = f"http://some_external_service_url.com/api/{urlencode(attributes, doseq=True)}" async with httpx.AsyncClient() as client: response = await client.get(url) return response.json().get("items")
如果连第三方依赖都不想装,自己实现一个带过期时间的全局字典存储也可以,每次取值时校验过期时间清理无效数据即可。
场景2:后续有多实例/分布式部署需求
可以用Python自带的sqlite文件数据库,不需要额外部署数据库服务,只需要建一张极简单的表存储request_id、attributes、expire_time三个字段即可,成本远低于单独搭Redis或者维护业务库表。
内容的提问来源于stack exchange,提问作者mrgrj
相关产品推荐
相关产品推荐

