asyncio event_loop in a Flask app

Question

What is the best method to have an asyncio event loop run in a Flask app?

My main.py looks like this:

if __name__ == '__main__':
    try:
        app.run(host='0.0.0.0', port=8000, debug=True)
    except:
        logging.critical('server: CRASHED: Got exception on main handler')
        logging.critical(traceback.format_exc())
        raise

To add the option of async tasks, I needed to create an event_loop before running the app, but even when stopping the app run, a background thread still hangs (observable in debugger)

if __name__ == '__main__':
    try:
        app.event_loop = asyncio.get_event_loop()
        app.run(host='0.0.0.0', port=8000, debug=True)
    except:
        logging.critical('server: CRASHED: Got exception on main handler')
        logging.critical(traceback.format_exc())
        raise
    finally:
        app.event_loop.stop()
        app.event_loop.run_until_complete(app.event_loop.shutdown_asyncgens())
        app.event_loop.close()

And using the following to create async tasks:

def future_callback(fut):
    if fut.exception():
        logging.error(fut.exception())

def fire_and_forget(func, *args, **kwargs):
    if callable(func):
        future = app.event_loop.run_in_executor(None, func, *args, **kwargs)
        future.add_done_callback(future_callback)
    else:
        raise TypeError('Task must be a callable')

The only solution I could find was to add exit() at the end of the finally block, but I don't think its the correct solution.

Answer 1

I initially took the approach of 1 event_loop for the entire flask app, but didn't like that for a few different reasons.

Instead, I created a helper file, custom_flask_async.py with 2 functions in,

import asyncio


def get_set_event_loop():
    try:
        return asyncio.get_event_loop()
    except RuntimeError as e:
        if e.args[0].startswith('There is no current event loop'):
            asyncio.set_event_loop(asyncio.new_event_loop())
            return asyncio.get_event_loop()
        raise e


def run_until_complete(tasks):
    return get_set_event_loop().run_until_complete(asyncio.gather(*tasks))

I only check for and get_set the event_loop if it's going to be used, and waited upon in a particular request. And because the main way I'm using this is with a run_until_complete with a gather, I've that in the helper too.

I'm also not sure if this is the best approach, or what drawbacks there are to this method, but it definitely works for my purpose.