如何在FastAPI中实现作为流量管理属性的服务超时？

方案一：通过Uvicorn服务器配置全局超时 #

Uvicorn作为FastAPI的ASGI服务器，本身提供连接层面的全局超时参数，适合统一设置连接生命周期的超时：

• timeout_keep_alive：保持连接的超时时间，默认5秒
• timeout_graceful_shutdown：服务优雅关闭的超时时间，默认10秒

修改启动代码即可生效：

from fastapi import FastAPI
from typing import Optional
import uvicorn

app = FastAPI()

class User(BaseModel):
    username: str

user_list = ["John", "Abrahim", "Kiana"]

@app.get('/user/{id}/')
def users(id:int):
    return user_list[id]

if __name__ == "__main__":
    uvicorn.run(
        app, 
        host="127.0.0.1",
        port=8000,
        timeout_keep_alive=10,  # 保持连接超时10秒
        timeout_graceful_shutdown=15  # 优雅关闭超时15秒
    )

方案二：单个路由的请求超时控制（异步场景） #

如果路由采用异步写法，可直接用asyncio.wait_for给单个请求设置超时，避免长时间阻塞：

from fastapi import FastAPI
from typing import Optional
import uvicorn
import asyncio

app = FastAPI()

user_list = ["John", "Abrahim", "Kiana"]

# 异步路由+超时控制
@app.get('/user/{id}/')
async def users(id: int):
    try:
        # 设置3秒超时，超时抛出TimeoutError
        result = await asyncio.wait_for(fetch_user(id), timeout=3)
        return result
    except asyncio.TimeoutError:
        return {"error": "请求超时，请稍后重试"}

async def fetch_user(id: int):
    # 模拟耗时业务操作（如数据库查询）
    await asyncio.sleep(4)  # 故意超过超时时间用于测试
    return user_list[id]

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

方案三：全局请求超时中间件 #

通过自定义中间件，为所有路由设置统一的请求超时，同步、异步路由均适用：

from fastapi import FastAPI, Request
from typing import Optional
import uvicorn
import asyncio
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

app = FastAPI()

user_list = ["John", "Abrahim", "Kiana"]

class TimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, timeout: int = 5):
        super().__init__(app)
        self.timeout = timeout

    async def dispatch(self, request: Request, call_next):
        try:
            return await asyncio.wait_for(call_next(request), timeout=self.timeout)
        except asyncio.TimeoutError:
            return JSONResponse(
                status_code=504,
                content={"error": "请求超时，服务器未能及时响应"}
            )

# 注册中间件，设置全局超时5秒
app.add_middleware(TimeoutMiddleware, timeout=5)

@app.get('/user/{id}/')
def users(id: int):
    # 模拟同步耗时操作
    import time
    time.sleep(6)  # 超过超时时间用于测试
    return user_list[id]

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

注意事项 #

• 同步路由的超时中间件仅会返回超时响应，但后台的同步阻塞任务可能仍会继续执行，若需彻底终止，需结合线程池封装同步逻辑。
• 建议同时配置服务器层面和请求层面的超时，形成多层超时保障。

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