How do I manage asynchronous startup and shutdown of resources in a Tornado application?

Question

I'd like to use aioredis in a Tornado application. However, I couldn't figure out a way to implement an async startup and shutdown of its resources since the Application class has no ASGI Lifespan events such as in Quart or FastAPI. In other words, I need to create a Redis pool before the app starts to serve requests and release that pool right after the app has finished or is about to end. The problem is that the aioredis pool creation is asynchronous, but the Tornado Application creation is synchronous.

The basic application looks like this:

    import os

from aioredis import create_redis_pool
from aioredis.commands import Redis
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application

from .handlers import hello

redis: Redis = None


async def start_resources() -> None:
    '''
    Initialize resources such as Redis and Database connections
    '''
    global redis
    REDIS_HOST = os.environ['REDIS_HOST']
    REDIS_PORT = os.environ['REDIS_PORT']
    redis = await create_redis_pool((REDIS_HOST, REDIS_PORT), encoding='utf-8')


async def close_resources() -> None:
    '''
    Release resources
    '''
    redis.close()
    await redis.wait_closed()


def create_app() -> Application:
    app = Application([
        ("/hello", hello.HelloHandler),
    ])

    return app


if __name__ == '__main__':

    app = create_app()
    http_server = HTTPServer(app)
    http_server.listen(8000)
    IOLoop.current().start()

It is important that I can use the startup and shutdown functions during tests too.

Any ideas?

Answer 1

The response from xyres is correct and put me on the right track. I only think it could be improved a little, so I am posting this alternative:

from contextlib import contextmanager

# ... previous code omitted for brevity


@contextmanager
def create_app() -> Application:
    IOLoop.current().run_sync(start_resources)
    try:
        app = Application([
            ("/hello", hello.HelloHandler),
        ])
        yield app
    finally:
        IOLoop.current().run_sync(close_resources)


if __name__ == '__main__':
    with create_app() as app:
        http_server = HTTPServer(app)
        http_server.listen(8000)
        IOLoop.current().start()

Also, to use this code in testing with pytest and pytest-tornado, you should create a conftest.py file like this:

from typing import Iterator

from pytest import fixture
from tornado.platform.asyncio import AsyncIOLoop
from tornado.web import Application

from app.main import create_app


@fixture
def app(io_loop: AsyncIOLoop) -> Iterator[Application]:
    '''
    Return a Tornado.web.Application object with initialized resources
    '''
    with create_app() as app:
        yield app

Note that it is important to declare io_loop as a dependency injection.