asyncio event_loop в приложении Flask

Question

Каков наилучший способ запуска цикла событий asyncio в приложении Flask?

Мой main.py выглядит так:

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

Чтобы добавить опцию асинхронных задач, мне нужно было создать event_loop перед запуском app, но даже при остановке запуска приложения фоновый поток все еще зависает (это можно наблюдать в отладчике)

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()

И используя следующее для создания асинхронных задач:

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')

Единственное решение, которое я смог найти, это добавить exit() в конце finally блок, но я не думаю, что это правильное решение.

Answer 1

Первоначально я использовал подход 1 event_loop для всего приложения фляги, но мне это не понравилось по нескольким причинам.

Вместо этого я создал вспомогательный файл, custom_flask_async.py с 2 функциями,

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

Я только проверяю и get_set event_loop, если он будет использоваться и ожидался в конкретном запросе. И потому, что основной способ, которым я использую это с run_until_complete с собранием, я тоже в помощнике.

Я также не уверен, является ли это лучшим подходом, или какие недостатки есть у этого метода, но он определенно работает для моей цели.

Answer 2

В конце концов, лучшее решение, которое я нашел, было разместить мое приложение Flask в uwsgi который позволяет использовать mules а также spooler для асинхронных задач

https://uwsgi-docs.readthedocs.io/en/latest/Spooler.html

https://uwsgi-docs.readthedocs.io/en/latest/Mules.html