如何让uvicorn运行异步构建的应用程序?
鉴于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?
回答
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并在其事件循环中以这种方式执行:
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)