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

你不用用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并在其事件循环中以这种方式执行：

 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)