You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI路由方法/端点之间传递数据的推荐实现方案

最优方案分场景选择,都可以规避你提到的额外组件开销问题

首先先明确完整请求链路的前提:第三方服务调用endpoint_1拿到唯一标识后,需要将该标识同步给前端,前端携带该标识调用endpoint_2才能完成数据匹配,这是跨独立请求传递数据的必要链路。

场景1:单实例部署、无分布式扩容需求

直接用内存级TTL缓存即可,不需要额外部署任何服务,实现成本最低。
你可以用纯Python实现的cachetools库的TTLCache,自带过期自动清理、容量上限控制,不会出现内存泄漏问题:

  1. 先安装依赖:
pip install cachetools
  1. 改造后的代码示例:
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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.30 04:18:01