如何在FastAPI视图中执行函数并立即返回HTTP响应（无需等待）

方法一：使用FastAPI内置的BackgroundTasks（推荐） #

FastAPI原生支持后台任务，这是官方推荐的实现方式，无需额外处理线程或事件循环的复杂逻辑。

修改后的代码如下：

from fastapi import FastAPI, Request, Form, BackgroundTasks  # 导入BackgroundTasks
import uvicorn
from testphoto.utils.logger import get_log
import datetime
import time
import asyncio

log = get_log()

app = FastAPI()

def process():
    log.info("Sleeping at "+str(datetime.datetime.now()))
    time.sleep(5)
    log.info("Woke up at "+str(datetime.datetime.now()))
    return "Success"

@app.post("/api/photos")
async def root(request: Request, photo: str = Form(), background_tasks: BackgroundTasks = None):
    background_tasks.add_task(process)  # 将process注册为后台任务
    return {"message": "Hello World"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8008)

修改后，请求会立即返回响应，process函数则在后台继续执行。

方法二：用asyncio线程池执行同步函数 #

如果不想依赖FastAPI的BackgroundTasks，也可以将同步的process函数放到asyncio的线程池中执行，避免阻塞当前请求的协程：

from fastapi import FastAPI, Request, Form
import uvicorn
from testphoto.utils.logger import get_log
import datetime
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor

log = get_log()

app = FastAPI()

# 创建线程池执行器
executor = ThreadPoolExecutor()

def process():
    log.info("Sleeping at "+str(datetime.datetime.now()))
    time.sleep(5)
    log.info("Woke up at "+str(datetime.datetime.now()))
    return "Success"

@app.post("/api/photos")
async def root(request: Request, photo: str = Form()):
    # 将process提交到线程池，不等待执行结果
    asyncio.get_event_loop().run_in_executor(executor, process)
    return {"message": "Hello World"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8008)

注意事项 #

• 上述两种方法的后台任务都依赖当前Uvicorn进程，若进程意外退出，未完成的任务会被终止。如果需要更可靠的后台任务（比如服务重启后仍能继续执行），建议使用消息队列架构（如Redis+Celery），但这属于更复杂的生产级方案。
• 不要在后台任务中直接操作请求相关对象（如request），请求返回后这些对象可能已被销毁，会引发异常。

内容的提问来源于stack exchange，提问作者Arka-cell