Uvicorn: `uvicorn.run` event loop behaviour when running programmatically.

[Gyllsdorff]

I am trying to run Uvicorn in the same asyncio loop as some other components like the asyncpg DB connection pool.

As far as I can tell we can specify the callable that receives the scope and creates the request coroutine two ways:

• Passing a str containing the path to uvicorn.run but then we can't setup our dependencies (db, redis) since that function is called each new request. We could load the dependencies by creating them when the file is imported but there is no way that will pass codereview.

• Pass a callable to uvicorn.run with the dependencies in the scope. Unfortunately uvicorn destroy the current event loop and replaces the event loop policy during the setup phase.

What is the recommended way to do things like this? If I could pass a loop instance in the loop parameter to run and have uvicorn use that directly without destroying it then it then I could use alternative 2 as long as we don't call unicorn.run from a async function due to the server.run behaviour..

@tomchristie

[tomchristie]

Perhaps we need a loop="no_setup" option?

Sets up the connection pool on the first incoming request. (An ASGI middleware could work there for example...)

[Gyllsdorff]

That would require the developer to know about a "magic" uvicorn setup function but that is a possible solution. Another is to just load and use the loop like is is done today if it is a string, otherwise just use the loop. I made a quick PR with the required code changes, PR #186.

Sets up the connection pool on the first incoming request.

I don't think that would be a good idea. :)

When a deploy is done the application should do sys.exit() if it can't reach the db, conf system, etc as soon as possible. Deferring the setup until the first request would also introduce race conditions and other problems since uvicorn would need to be aware that the first request triggers a db connection pool creation, logger configuration and configuration fetching and block all other requests until that first request is done.

[tomchristie]

I don't think that would be a good idea.

Agreed. You can block all incoming requests against a semaphore until startup is complete, but it makes much more sense to run all setup to completion first.

[tomchristie]

I am trying to run Uvicorn in the same asyncio loop as some other components like the asyncpg DB connection pool.

Unrelated aside - what are you running Uvicorn against? Channels, APIStar, Starlette, or a plain ASGI app?

Something else that'd help with this issue will be if/when ASGI gets startup/shutdown messaging. Eg. see discussion on https://github.com/django/asgiref/pull/61, since we'll then be able to notify applications on startup, and any events such as connection pool setup can happen then. (Also, making sure to halt immediately if an exception is raised during startup)

[alexshpilkin]

Am I correct in assuming this is no longer supported? As far as I can see, Config.setup_event_loop (config.py line 213) now once again unconditionally dereferences LOOP_SETUPS with the loop argument.

[Eduard-gan]

Pull request was merged and then the behavior was removed i believe in this commit: https://github.com/encode/uvicorn/pull/276

In my case i need to run message-consumers based on aio-pika aside a set HTTP endpoints served by uvicorn in the same event loop in my microservice application.

The only way of doing this i found is to move configuration and initialization of my consumers inside app served by uvicorn.

for example like this(sorry for a shitcode but the way is exactly the same - we're just not messing around with event loop before uvicorn kiks in):

import uvicorn
import json
import asyncio
from aio_pika import connect_robust


async def consume_qq(channel):
    qq = await channel.declare_queue("qq")
    async with qq.iterator() as queue_iter:
        async for message in queue_iter:
            await kek(message=message)


async def queues_consuming_app():
    connection = await connect_robust(host="localhost")
    channel = await connection.channel()
    print(f"Got connection {connection}")
    print(f"Got channel {channel}")

    # Consumer-level error-handling
    while True:
        try:
            print("Starting to consume QQ")
            await consume_qq(channel=channel)
        except Exception as e:
            print(f"CRITICAL ERROR IN CUNSUMER OF QQ: [ {e} ]")
            await asyncio.sleep(7)


async def kek(message=None, text=None):
    await asyncio.sleep(3)
    if message:
        async with message.process(requeue=True, reject_on_redelivered=True):
            print(f'Got message: {json.loads(getattr(message, "body", b"{}").decode())}')
    else:
        print(f"No messege, text is {text}")


async def root_app(scope, receive, send):
    print(f"HTTP CALL OR LIFESPAN: {scope}, {receive}, {send}")
    # Hooking up message consuming
    asyncio.ensure_future(queues_consuming_app(), loop=asyncio.get_event_loop())


if __name__ == "__main__":
    uvicorn.run(root_app, host="127.0.0.1", port=5000, log_level="info")