如何在FastAPI中实现数据库异步通信与并发任务执行？

无需Celery，FastAPI自带异步机制即可实现 #

你要的并行执行写入、读取数据库的操作，完全不需要引入Celery这类任务队列框架，直接用Python的asyncio配合FastAPI就能搞定。

具体实现方式 #

Python的asyncio.gather()可以让多个协程并行执行，把写入和读取操作放进这个方法里，它们就会同时运行，而非串行等待。

先写一个异步读取函数示例：

async def get_latest_summaries(limit: int = 5) -> list[TextSummary]:
    return await TextSummary.all().limit(limit)

然后在接口中并行执行两个异步任务：

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.post("/summaries")
async def create_summary(payload: SummaryPayloadSchema):
    # 定义两个异步任务
    write_task = post(payload)
    read_task = get_latest_summaries()
    
    # 等待两个任务同时完成
    summary_id, latest_summaries = await asyncio.gather(write_task, read_task)
    
    return {
        "created_summary_id": summary_id,
        "latest_summaries": latest_summaries
    }

这样数据库的写入和读取请求会同时发起，充分利用异步IO的优势，不会互相阻塞等待。

同一个端点异步获取其他记录，完全适用该方案 #

不管是在同一个端点里并行执行写入+读取，还是多个读取操作，只要都是异步函数，都可以用asyncio.gather()实现并行。需要注意：你的数据库驱动必须是异步版本（比如用asyncpg操作PostgreSQL、motor操作MongoDB），如果用同步驱动，就算用了asyncio也没法真正并行，反而会阻塞事件循环。

什么时候需要Celery？ #

只有当你要处理超长耗时任务（比如生成大文件、调用响应极慢的第三方接口），或者需要离线执行、定时执行任务时，才需要Celery这类任务队列。像你现在这种轻量数据库读写操作，FastAPI自身的异步机制完全够用。

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