使用 FastAPI 与异步循环交互

Question

我正在运行“workers”协程，其工作是等待 5 秒，从 asyncio.Queue() 获取值并不断打印出来。

q = asyncio.Queue()

def worker():
    while True:
        await asyncio.sleep(5)
        i = await q.get()
        print(i)
        q.task_done()

async def main(q):
    workers = [asyncio.create_task(worker()) for n in range(10)]
    await asyncio.gather(*workers)


if __name__ == "__main__":
    asyncio.run(main())

我希望能够使用 FastAPI 通过 http 请求与队列进行交互。例如，POST 请求会将项目“放入”队列中以供工作人员打印。

我不确定如何与 FastAPI 同时运行工作线程的协程来实现这种效果。我相信 Uvicorn 有自己的事件循环，并且我尝试使用 asyncio 方法没有成功。

我认为路由器看起来像这样。

@app.post("/")
async def put_queue(data:str):
    return q.put(data)

我希望有一些东西会产生这样的效果：

await asyncio.gather(main(),{FastApi() app run})

Answer 1

一种选择是添加一个任务，将主协程包装在启动事件中

import asyncio
@app.on_event("startup")
async def startup_event():
    asyncio.create_task(main())

这将在应用程序完全启动之前安排您的主协程。

这里重要的是，您不要等待创建的任务，因为它基本上会永远阻止startup_event