Как включить uvicorn для запуска асинхронно созданного приложения?

#python #async-await #python-asyncio #uvicorn #asgi

#python #асинхронное ожидание #python-asyncio #uvicorn #asgi

Вопрос:

Учитывая 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 ?

Ответ №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 и выполнен таким образом в его eventloop:

  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)