如何用FastAPI搭建无需等待响应即可触发Python文件执行查询的API

FastAPI 自带 BackgroundTasks 组件，可以将耗时任务放到后台异步执行，接口无需等待任务完成即可直接返回响应，具体实现代码如下：

完整修改后的代码 #

from fastapi import BackgroundTasks, FastAPI, HTTPException, Depends
# 你原有的其他导入保持不变，比如 details、oath2_scheme 等

app = FastAPI()

# 封装你的耗时查询逻辑，也可以直接把原逻辑写到add_task参数里
def run_query_task(database: str, entity_name: str):
    # 这里执行你的实际耗时查询逻辑
    value = details.get_query_status(database, entity_name)
    # 你可以在这里加后续的结果处理逻辑，比如存库、发通知等，不需要返回给接口调用方
    # 注意原代码的逻辑错误：如果value为None时取value[1]会报错，这里调整逻辑
    if value is not None:
        # 可选操作：把结果存到Redis/数据库供后续查询
        # redis.set(f"query_result:{database}:{entity_name}", value)
        pass
    else:
        # 可选操作：记录错误日志
        # logger.error(f"查询失败：database={database}, entity_name={entity_name}")
        pass

@app.get("/trigger_query")
def trigger_query(
    database: str, 
    entity_name: str, 
    token: str = Depends(oath2_scheme),
    background_tasks: BackgroundTasks
):
    # 将耗时任务添加到后台队列
    background_tasks.add_task(run_query_task, database, entity_name)
    # 直接返回触发成功的结果，不需要等待查询执行完成
    return {'status_code': 200, 'detail': "查询任务已触发"}

注意事项 #

• BackgroundTasks 适合轻量级的异步任务场景，如果你的查询任务很重、需要支持失败重试、任务持久化，建议接入 Celery、RQ 等专业的分布式任务队列
• 如果你封装的任务函数是async def定义的异步函数，BackgroundTasks 也可以直接支持，不需要额外修改调用逻辑
• 如果需要后续获取任务执行结果，可以生成唯一的任务ID返回给调用方，同时把任务状态和结果存储在 Redis、数据库等公共存储中，额外开发一个查询任务状态的接口即可
• 原代码存在逻辑错误：当 value is None 时尝试访问 value[1] 会抛出索引异常，使用时需要调整对应的判断逻辑

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