FastAPI应用为何随机返回‘Failed to fetch. Possible Reasons:’错误？

你的判断没错，这个问题确实和接口运行时长超过默认超时限制有关，再加上代码里的同步阻塞调用放大了问题，具体原因和解决方法如下：

核心原因 #

1. 同步阻塞调用卡住事件循环：在async定义的接口函数里使用time.sleep()是同步阻塞操作，会完全占用FastAPI/Uvicorn的事件循环，导致整个服务无法处理其他请求，同时长时间的阻塞会触发超时机制。
2. 默认超时限制：Uvicorn、浏览器或者前端反向代理（如Nginx）都有默认的超时配置（一般在10分钟以内），当请求持续时间超过这个阈值，就会被强制断开连接，返回你看到的超时错误。

具体解决步骤 #

1. 替换同步阻塞为异步调用 #

首先把time.sleep()换成异步的asyncio.sleep()，避免阻塞事件循环，修改后的代码如下：

from fastapi import FastAPI
import uvicorn
import asyncio  # 新增导入
from loguru import logger
import sys

logger.add("file_{time}.log")
logger.add(sys.stdout, colorize=True, format="<green>{time}</green> <level>{message}</level>")

app = FastAPI()

@app.get("/")
async def root():
    logger.info("starting SLEEP loop")
    for i in range(0, 63):
        await asyncio.sleep(10)  # 替换为异步sleep
        logger.info(f"slept for {(i+1)*10} seconds or {float((i+1)*10/60)} minutes")

    return {"message": "Hello World"}

uvicorn.run(app)

2. 调整Uvicorn超时参数 #

如果必须让请求持续10分钟以上，需要在启动Uvicorn时调整超时相关参数，比如：

uvicorn.run(
    app,
    timeout_keep_alive=630,  # 保持连接超时设置为10分30秒，覆盖默认的60秒
    timeout_graceful_shutdown=630  # 优雅关闭超时也对应调整
)

或者用命令行启动的话：

uvicorn main:app --timeout-keep-alive 630 --timeout-graceful-shutdown 630

3. 优化架构：用后台任务处理长耗时操作 #

HTTP请求天生不适合长时间保持连接，更合理的方式是把长耗时任务放到后台执行，给客户端返回任务ID，之后让客户端通过轮询获取结果，示例代码如下：

from fastapi import FastAPI, BackgroundTasks
import asyncio
from loguru import logger
import sys
from uuid import uuid4

logger.add("file_{time}.log")
logger.add(sys.stdout, colorize=True, format="<green>{time}</green> <level>{message}</level>")

app = FastAPI()
task_results = {}  # 生产环境建议用Redis等持久化存储替代内存字典

async def long_running_task(task_id: str):
    logger.info(f"starting long task {task_id}")
    for i in range(0, 63):
        await asyncio.sleep(10)
        logger.info(f"task {task_id} slept for {(i+1)*10} seconds")
    task_results[task_id] = {"message": "Hello World"}

@app.get("/start-task")
async def start_task(background_tasks: BackgroundTasks):
    task_id = str(uuid4())
    background_tasks.add_task(long_running_task, task_id)
    return {"task_id": task_id, "status": "running"}

@app.get("/check-task/{task_id}")
async def check_task(task_id: str):
    if task_id not in task_results:
        return {"task_id": task_id, "status": "running"}
    return {"task_id": task_id, "status": "completed", "result": task_results.pop(task_id)}

额外注意事项 #

• 如果你的服务前端还有Nginx等反向代理，需要同步调整Nginx的proxy_connect_timeout、proxy_send_timeout、proxy_read_timeout参数，设置为大于任务执行的时长。
• 生产环境中不要用内存字典存储任务结果，建议使用Redis、数据库等持久化存储方案，避免服务重启后数据丢失。

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