如何在Python中为GraphQL FastAPI API设置超时后终止请求

核心思路 #

利用asyncio.wait_for()给异步任务设置超时阈值，当任务超过指定时长未完成时，主动抛出超时错误并返回给客户端，避免请求无限挂起。

适配GraphQL场景的完整实现 #

以常用的ariadne GraphQL库结合FastAPI为例，代码如下：

from fastapi import FastAPI, HTTPException
from ariadne import QueryType, make_executable_schema
from ariadne.asgi import GraphQL
import asyncio

# 定义GraphQL Schema
type_defs = """
    type Query {
        fetchData(input: String!): String
    }
"""

query = QueryType()

# 模拟耗时数据库查询（替换为实际异步数据库操作）
async def long_running_db_query(input_param: str):
    await asyncio.sleep(20)  # 模拟20秒耗时任务
    return f"处理完成：{input_param}"

@query.field("fetchData")
async def resolve_fetch_data(*_, input: str):
    try:
        # 设置15秒超时
        result = await asyncio.wait_for(long_running_db_query(input), timeout=15)
        return result
    except asyncio.TimeoutError:
        raise HTTPException(status_code=408, detail="请求超时，请稍后重试")

# 挂载GraphQL端点到FastAPI
schema = make_executable_schema(type_defs, query)
app = FastAPI()
app.mount("/graphql", GraphQL(schema, debug=True))

关键注意事项 #

• 必须确保数据库查询是异步实现（比如用asyncpg操作PostgreSQL、motor操作MongoDB），否则asyncio.wait_for()无法中断阻塞的同步任务。
• 超时时间建议根据业务场景设置在10-15秒，平衡用户体验和业务需求。
• 如果使用其他GraphQL库（如strawberry），核心逻辑一致：在Resolver函数中用asyncio.wait_for()包裹耗时异步任务。

普通FastAPI端点的超时实现（参考示例） #

如果是REST端点，可补充完善你的示例代码如下：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio

app = FastAPI()

class XYZRequest(BaseModel):
    input_param: str

# 模拟耗时任务
async def long_running_task(input: str):
    await asyncio.sleep(8)  # 模拟耗时操作
    return f"结果：{input}"

@app.post("/xyz")
async def xyz(xyz_input: XYZRequest):
    try:
        # 设置5秒超时
        result = await asyncio.wait_for(long_running_task(xyz_input.input_param), timeout=5)
        return {"result": result}
    except asyncio.TimeoutError:
        raise HTTPException(status_code=408, detail="请求超时")

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