This is a followup of the previous article on using Starlette together with Python-SocketIO and background processes.

I’ve made some improvements, especially to handle SIGINT (ctrl+c) properly. Additionally, I wanted to start background processes without waiting for a client to connect (this is a solution I saw online).

The core problem was that all async tasks should be on the same loop. The solution took some digging around uvicorn internals, but the following worked:

  • Get the Uvicorn server as an awaitable;
  • Join the awaitable with your background tasks and return when the first completes
  • Run the composed application.

This properly responds to SIGINT and all processes start without waiting for outside interaction.

Background tasks are more like workers here. One-off tasks fall in two categories:

  • Startup or shutdown: Use Starlette hooks for startup and shutdown to handle work there;
  • Based on interactions from outside: Just call tasks when an API is called.
import logging
import uvicorn
from app import app
import asyncio
from uvicorn.loops.uvloop import uvloop_setup


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)-15s %(levelname)-8s %(message)s"
)


def uvicorn_task():
    """ Returns running the app in uvicorn as an awaitable to join the main
        asyncio loop. """
    config = uvicorn.Config(app, host='0.0.0.0', port=8000)
    server = uvicorn.Server(config)

    return server.serve()


async def main(app):
    """ Joins unicorn together with background tasks defined in the app. """
    # Make the list with awaitables
    aws = [uvicorn_task(),
           *app.background_tasks()]
    # Run and return when the first completes or is cancelled
    await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)


if __name__ == '__main__':
    # Set up the loop
    uvloop_setup()
    loop = asyncio.get_event_loop()

    # Run the main loop
    asyncio.run(main(app))