How to use event loop created by uvicorn?

user_8065970 0
Viewed 2504

I am using uvicorn and I need to use the existing event loop. I'm using the following command:

loop = asyncio.get_event_loop()

But when I use this line, the code get stuck. But if I use new_event_loop, I get the following error:

RuntimeError: Task <Task pending coro=<open_connection() running at /usr/lib/python3.6/asyncio/streams.py:81> cb=[_release_waiter(<Future pendi...sk._wakeup()]>)() at /usr/lib/python3.6/asyncio/tasks.py:316]> got Future <Future pending> attached to a different loop

Please help me to understand how to use the existing loop created by uvicorn.

python-asyncio python
1 Answers

You can read this problem @ https://github.com/encode/uvicorn/issues/706

Basically you have to create your own event loop and pass it to uvicorn.

import asyncio
from uvicorn import Config, Server


async def app(scope, 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!',
    })


loop = asyncio.new_event_loop()

config = Config(app=app, loop=loop)
server = Server(config)
loop.run_until_complete(server.serve())
user_9023840
Related