python - 如何让 uvicorn 运行异步构建的应用程序？

Question

给定 main.py :

import asyncio

async def new_app():
    # Await some things.

    async def app(scope, receive, send):
        ...

    return app

app = asyncio.run(new_app())

其次是:

uvicorn main.app

给出:

RuntimeError: asyncio.run() cannot be called from a running event loop

这是因为 uvicorn在导入我的应用程序之前已经启动了一个事件循环。如何在uvicorn下异步构建应用程序?

Answer 1

Yoe 不需要使用 asyncio.run .你的类或函数应该只实现 ASGI 界面。像这样，最简单的可行:

# main.py
def app(scope):
    async def asgi(receive, send):
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [[b"content-type", b"text/plain"]],
            }
        )
        await send({"type": "http.response.body", "body": b"Hello, world!"})

    return asgi

你可以在 uvicorn 下启动它:uvicorn main:app .
参数 main:app将由 uvicorn 解析导入和 executed在其事件循环中以这种方式:

 app = self.config.loaded_app
 scope: LifespanScope = {
     "type": "lifespan",
     "asgi": {"version": self.config.asgi_version, "spec_version": "2.0"},
 }
 await app(scope, self.receive, self.send)

如果你想制作一个可执行模块，你可以这样做:

import uvicorn
# app definition
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)