skip to Main Content

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?

2

Answers


  1. Chosen as BEST ANSWER

    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. 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