Асинхронный Flask 2.0

Flask 2.0 вышел 11 мая 2021 г. и добавляет встроенную поддержку асинхронных маршрутов (routes), обработчиков ошибок (error handlers), функций до (before request) и после (after request) запроса, а также обратных вызовов (коллбэков) разрыва!

Рассмотрим новые асинхронные функции Flask 2.0 и способы их использования в проектах с Flask.

Flask 2.0 async

Начиная с Flask 2.0, появляется возможность создавать обработчиков асинхронных маршрутов (routes), используя async/await:

import asyncio


async def async_get_data():
    await asyncio.sleep(1)
    return 'Done!'


@app.route("/data")
async def get_data():
    data = await async_get_data()
    return data

Асинхронные маршруты не сложнее в разработке, чем синхронные:

1. Просто устанавливаем Flask с async с помощью pip install "flask[async]".

2. Затем можно добавлять ключевое слово async в функции и использовать await.

Как это все работает?

На следующей схеме представлено исполнение асинхронного кода в Flask 2.0:

Чтобы запускать асинхронный код на Python, необходим цикл с ожиданием событий (event loop) и запуском сопрограмм (coroutines). Flask 2.0 берет на себя включение асинхронного цикла с ожиданием событий (как обычно, это делается через asyncio.run()) для запуска сопрограмм.

При обработке функции маршрута async появляется новый подпоток (sub-thread). Внутри этого подпотока будет выполняться цикл с ожиданием событий для запуска обработчика маршрута (сопрограммы).

В данной реализации используется библиотека asgiref (если точнее, функционал asynctosync), используемая в django для выполнения асинхронного кода.

Подробнее о реализации можно посмотреть в документации по async_to_sync() в исходном коде Flask.

Данная реализация хороша тем, что она позволяет запускать Flask с любым типом процесса-исполнителя (worker) (потоки, gevent, eventlet и т.д.).

До появления Flask 2.0 для выполнение асинхронного кода требовалось делать новые асинхронные циклы с ожиданием событий в каждом обработчике маршрута, что требовало запуска приложения Flask с использованием потоковых процессов-исполнителей. Подробнее далее...

Кроме того, использование асинхронных обработчиков маршрутов дает обратную совместимость. Можно как угодно сочетать обработчики асинхронных и синхронных маршрутов в приложении Flask без каких-либо компромиссов в части производительности. Это позволяет сразу перейти к созданию прототипа даже одного обработчика асинхронного маршрута в существующем проекте Flask.

Почему не нужен ASGI?

Flask изначально создавался как синхронный веб-фреймворк, реализующая протокол WSGI (Интерфейс веб-серверного шлюза).

Во Flask каждый запрос обрабатывается индивидуально в процессе-исполнителе. Асинхронная функциональность, добавленная во Flask 2.0, всегда находится в пределах одного обрабатываемого запроса:

Не забывайте, что даже невзирая на возможность исполнения асинхронного кода во Flask, его исполнение осуществляется в контексте синхронного фреймворка. Другими словами, пусть даже можно исполнять разные асинхронные задачи в одном запросе (request), каждая асинхронная задача должна завершиться до отправки ответа обратно. Поэтому не во всех ситуациях асинхронные маршруты будут по-настоящему выгодны. Есть и другие веб-фреймворки Python, в которых поддерживается ASGI (Интерфейс асинхронного серверного шлюза) с поддержкой асинхронных стеков вызовов, что позволяет исполнять маршруты параллельно:

Фреймворк	Асинхронный стек запроса (например, поддержка ASGI)	Асинхронные маршруты
Quart	Да	Да
Django >= 3.2	Да	Да
FastAPI	Да	Да
Flask >= 2.0	Нет	Да

Когда следует использовать async?

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

Оно идеально подходит для операций, ориентированных на ввод/вывод, когда выполняются оба следующих условия:

1. Есть ряд операций

2. Полное исполнение каждой операции занимает менее нескольких секунд

Например:

1. Запросы по HTTP или API

2. Взаимодействие с базой данных

3. Работа с файловой системой

Оно не подходит для фоновых и долгосрочных задач, а также для операций, связанных с работой процессора, например:

1. Запуск моделей машинного обучения

2. Обработка изображений или PDF

3. Выполнение резервного копирования

Такие задачи лучше реализовывать с помощью очереди задач (task queue), например Celery, для управления отдельными длительными задачами.

Асинхронные HTTP-вызовы

Асинхронный подход действительно приносит дивиденды, когда нужно сделать ряд HTTP-запросов на внешний веб-сайт или API. Каждому запросу потребуется значительное время на получение ответа. Вследствие данного времени ожидания пользователи посчитают веб-приложение медленным или инертным.

Вместо того, чтобы делать внешние запросы по одному (через пакет requests), можно значительно ускорить процесс, используя async/await.

В синхронном подходе выполняется внешний вызов API (например, через GET), а затем приложение ожидает ответа. Период ожидания ответа называется задержкой (latency), которая зависит от качества подключения к Интернету и времени отклика сервера. В данном случае задержка, возможно, составит около 0,2–1,5 секунды на запрос.

В асинхронном подходе выполняется внешний вызов API, а затем обработка продолжается переходом к следующему вызову API. Полученный от внешнего сервера ответ сразу обрабатывается. Так ресурсы задействуются гораздо эффективнее.

В целом, асинхронное программирование идеально подходит для ситуаций наподобие вышеописанной, когда выполняется несколько внешних вызовов и приходится подолгу ждать ответов из интерфейса ввода/вывода.

Обработчик асинхронных маршрутов

aiohttp – пакет, в котором asyncio (асинхронный ввод/вывод) используется для создания асинхронных http-клиентов и серверов. Для тех, кто знаком с пакетом requests для синхронного выполнения HTTP-вызовов, aiohttp будет аналогичным пакетом, ориентированным на асинхронные HTTP-вызовы.

Рассмотрим пример использования aiohttp в маршруте Flask:

urls = ['https://www.kennedyrecipes.com',
        'https://www.kennedyrecipes.com/breakfast/pancakes/',
        'https://www.kennedyrecipes.com/breakfast/honey_bran_muffins/']

# вспомогательные функции

async def fetch_url(session, url):
    """извлечь указанный URL-адрес, используя указанную сессию aiohttp."""
    response = await session.get(url)
    return {'url': response.url, 'status': response.status}


# маршруты

@app.route('/async_get_urls_v2')
async def async_get_urls_v2():
    """асинхронно извлекаем список url-адресов."""
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.create_task(fetch_url(session, url))
            tasks.append(task)
        sites = await asyncio.gather(*tasks)

    # Generate the HTML response
    response = '<h1>URLs:</h1>'
    for site in sites:
        response += f"<p>URL: {site['url']} --- Status Code: {site['status']}</p>"

    return response

Исходный код данного примера есть в репозитории flask-async на gitlab.

В сопрограмме async_get_urls_v2() используется стандартный паттерн asyncio:

1. Создаем несколько асинхронных задач (asyncio.create_task())

2. Запускаем их одновременно (asyncio.gather())

Тестирование асинхронных маршрутов

Можно протестировать обработчика асинхронного маршрута по стандартному подходу с pytest, так как Flask берет на себя всю асинхронную обработку:

@pytest.fixture(scope='module')
def test_client():
    # создаем тестового клиента, используя приложение flask
    with app.test_client() as testing_client:
        yield testing_client  # именно здесь происходит тестирование!


def test_async_get_urls_v2(test_client):
    """
    ДАНО: тестовый клиент Flask
    ЕСЛИ: идет запрос на страницу '/async_get_urls_v2' (GET)
    ТО: проверить правильность ответа
    """
    response = test_client.get('/async_get_urls_v2')
    assert response.status_code == 200
    assert b'URLs' in response.data

Это будет базовая проверка на верность ответа с URL-адреса /async_get_urls_v2 через зафиксированный объект (fixture) test_client.

Еще примеры с async

Обратные вызовы запроса тоже может выполнять асинхронно во Flask 2.0:

# вспомогательные функции

async def load_user_from_database():
    """имитирует длительную операцию по загрузке пользователя из внешней базы данных."""
    app.logger.info('загрузка пользователя из базы данных...')
    await asyncio.sleep(1)


async def log_request_status():
    """имитирует длительную операцию по регистрации статуса запроса."""
    app.logger.info('регистрация статуса запроса...')
    await asyncio.sleep(1)


# обратные вызовы запроса

@app.before_request
async def app_before_request():
    await load_user_from_database()


@app.after_request
async def app_after_request(response):
    await log_request_status()
    return response

Аналогично с обработчиками ошибок:

# вспомогательные функции

async def send_error_email(error):
    """имитирует длительную операцию по регистрации ошибки."""
    app.logger.info('регистрация статуса ошибки...')
    await asyncio.sleep(1)


# обработчики ошибок

@app.errorhandler(500)
async def internal_error(error):
    await send_error_email(error)
    return '500 error', 500

Async во Flask 1.x

При работе с Flask 1.x можно имитировать поддержку async из Flask 2.0, используя asyncio.run() для управления асинхронным циклом с ожиданием событий:

# вспомогательные функции

async def fetch_url(session, url):
    """извлекаем указанный URL-адрес через указанную сессию aiohttp."""
    response = await session.get(url)
    return {'url': response.url, 'status': response.status}


async def get_all_urls():
    """асинхронно извлекаем список url-адресов с помощью aiohttp."""
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            task = asyncio.create_task(fetch_url(session, url))
            tasks.append(task)
        results = await asyncio.gather(*tasks)

    return results


# маршруты

@app.route('/async_get_urls_v1')
def async_get_urls_v1():
    """асинхронно извлекаем список url-адресов (работает во Flask 1.1.x при использовании потоков)."""
    sites = asyncio.run(get_all_urls())

    # генерируем ответ html
    response = '<h1>URLs:</h1>'
    for site in sites:
        response += f"<p>URL: {site['url']} --- Status Code: {site['status']}</p>"
    return response

Функциональность, которая реализуется в сопрограмме get_all_urls(), похожа на ту, которую мы рассмотрели для обработчика маршрута async_get_urls_v2().

Как это все работает?

Асинхронный цикл с ожиданием событий будет правильно работать во flask 1.x, если запускать приложение Flask с использованием потоков (именно они являются процессами-исполнителями по умолчанию в gunicorn, uWSGI и в сервере разработки Flask):

Каждый поток будет запускать экземпляр приложения Flask при обработке запроса. Внутри каждого потока создается отдельный асинхронный цикл с ожиданием событий для выполнения любых асинхронных операций.

Тестирование сопрограмм

Вот так можно использовать pytest-asyncio для тестирования асинхронного кода:

@pytest.mark.asyncio
async def test_fetch_url():
    """
    ДАНО: "асинхронный цикл с ожиданием событий
    ЕСЛИ: вызывается сопрограмма fetch_url()
    ТО: проверить правильность ответа
    """
    async with aiohttp.ClientSession() as session:
        result = await fetch_url(session, 'https://www.kennedyrecipes.com/baked_goods/bagels/')

    assert str(result['url']) == 'https://www.kennedyrecipes.com/baked_goods/bagels/'
    assert int(result['status']) == 200

В данной тестовой функции используется декоратор @pytest.mark.asyncio, который указывает pytest на необходимость исполнить сопрограмму как асинхронную задачу через асинхронный цикл с ожиданием событий.

Заключение

Во Flask 2.0 добавлена очень удачная поддержка асинхронного функционала. Однако асинхронный код следует использовать только, если он дает преимущество перед эквивалентным синхронным кодом. Мы увидели, что одним из примеров осмысленного использования асинхронного исполнения является ситуация, когда нужно сделать несколько HTTP-вызовов в обработчике маршрута.

Я сделал несколько тестов тайминга асинхронной функции из Flask 2.0 (async_get_urls_v2()) в сравнении с эквивалентной синхронной функцией. Выполнено по десять вызовов на каждый маршрут:

Тип	Среднее время (секунды)	Медианное время (секунды)
Синхронный	4,071443	3,419016
Асинхронный	0,531841	0,406068

Асинхронная версия быстрее примерно в 8 раз! Итак, если вам нужно сделать несколько внешних HTTP-вызовов в обработчике маршрута, повышенная сложность использования asyncio и aiohttp определенно оправдана вследствие значительного сокращения срока исполнения.

Если хотите лучше изучить Flask, обязательно посмотрите мой курс: Developing Web Applications with Python and Flask.