Question posted in
The official documentation can be found here.

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

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?

Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    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.


  2. 0

    To create the pool, call your coroutine using run_sync before you start the loop:

    if __name__ == '__main__':
    IOLoop.current().run_sync(start_resources)
    ...

    To destroy the pool before the program exits, use a try...finally block so that abrupt exits due to unhandled exceptions are also accounted for:

    if __name__ == '__main__': 
    # create db pool
    IOLoop.current().run_sync(start_resources)

    ...

    try:
        # start the loop
        IOLoop.current().start()
    except:
        pass
    finally:
        # this will close the pool before exiting
        IOLoop.current().run_sync(close_resources)
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search