Сообщить об ошибке.

Подписка на сигналы в веб-приложении Flask.

Содержание:


Что такое сигналы Flask?

Сигналы Flask помогают отделить приложения, отправляя уведомления, когда действия происходят в другом месте основной платформы или в других расширениях Flask. Короче говоря, сигналы позволяют определенным отправителям уведомлять подписчиков о том, что что-то произошло.
Поддержка сигналов обеспечивается сторонним модулем blinker и будет изящно отменена, если он не установлен.
Flask поставляется с двумя настроенными сигналами, другие расширения могут предоставить больше сигналов. Также имейте в виду, что сигналы предназначены для уведомления подписчиков и не должны побуждать подписчиков изменять данные. Есть сигналы, которые, например делают то же самое, что и некоторые из встроенных декораторов (например: сигнал flask.request_started() очень похож на декоратор @app.before_request()), но есть различия в том, как они работают. Например, основной обработчик @app.before_request() выполняется в определенном порядке и может преждевременно прервать запрос, вернув ответ. Напротив, все сигналы выполняются в неопределенном порядке и не изменяют никаких данных.
Большим преимуществом сигналов перед обработчиками является то, что можно безопасно подписаться на них всего за несколько строк кода. Эти временные подписки полезны, например, для модульного тестирования. Допустим, необходимо знать, какие шаблоны были отрисованы как часть запроса: сигналы позволяют сделать именно это.

Подписка на сигналы Flask.

Чтобы подписаться на сигнал, необходимо использовать метод сигнала Signal.connect(). Первый аргумент метода - это функция, которая должна вызываться при передаче сигнала, второй необязательный аргумент указывает отправителя. Чтобы отказаться от подписки на сигнал, можно использовать метод Signal.disconnect().
Для всех основных сигналов Flask, отправителем является приложение, которое отправило сигнал. Если не хотите прослушивать сигналы от всех приложений, то когда подписываетесь на сигнал, обязательно указывайте отправителя. Это особенно важно, если разрабатываете расширение.
Пример, вспомогательного менеджера контекста, который можно использовать в модульном тесте, чтобы определить, какие шаблоны были отрисованы и какие переменные были переданы в шаблон:
from flask import template_rendered
from contextlib import contextmanager

@contextmanager
def captured_templates(app):
    recorded = []
    def record(sender, template, context, **extra):
        recorded.append((template, context))
    # подписываемся на сигнал
    template_rendered.connect(record, app)
    try:
        yield recorded
    finally:
        # отменяем подписку на сигнал
        template_rendered.disconnect(record, app)
Теперь это можно легко связать с тестовым клиентом:
with captured_templates(app) as templates:
    rv = app.test_client().get('/')
    assert rv.status_code == 200
    assert len(templates) == 1
    template, context = templates[0]
    assert template.name == 'index.html'
    assert len(context['items']) == 10
Обязательно подписывайтесь с дополнительным аргументом **extra, чтобы вызовы не завершились ошибкой, если Flask введет новые аргументы в API сигналов.
Вся отрисовка шаблона в коде, выданном приложением app в теле блока with, теперь будет записана в переменной templates. Каждый раз, когда отображается шаблон, к нему добавляются объект шаблона, а также контекст.
Кроме того, существует удобный вспомогательный метод (Signal.connected_to()), который позволяет временно подписать функцию на сигнал с помощью диспетчера контекста самостоятельно. Поскольку возвращаемое значение диспетчера контекста не может быть указано таким образом, вы должны передать список в качестве аргумента:
from flask import template_rendered

def captured_templates(app, recorded, **extra):
    def record(sender, template, context):
        recorded.append((template, context))
    return template_rendered.connected_to(record, app)
Приведенный выше пример будет выглядеть так:
templates = []
with captured_templates(app, templates, **extra):
    ...
    template, context = templates[0]

Создание и отправка сигналов Flask.

Если нужно использовать сигналы в собственном приложении, то можно использовать сторонний модуль blinker напрямую. Чаще всего используются именованные сигналы в настраиваемом пространстве имен (в большинстве случаев это рекомендуется):
from blinker import Namespace
my_signals = Namespace()
Теперь можно создавать новые сигналы:
model_saved = my_signals.signal('model-saved')
Здесь, имя сигнала делает его уникальным, а также упрощает отладку. Можно получить доступ к имени сигнала с помощью атрибута model_saved.name.
Для разработчиков расширений: Если необходимо аккуратно выполнить отключение сигналов из-за отсутствия установки модуля blinker, то это можно сделать с помощью класса flask.signals.Namespace.

Отправка сигналов.

Если нужно послать сигнал, то можно сделать это, вызвав метод Signal.send(). Он принимает отправителя в качестве первого аргумента и, возможно, некоторые ключевые аргументы, которые пересылаются подписчикам сигнала:
class Model(object):
    ...

    def save(self):
        model_saved.send(self)
Старайтесь всегда выбирать хорошего отправителя. Если есть класс, который посылает сигнал, передайте self в качестве отправителя. Если сигнал посылается случайной функцией, то в качестве отправителя можно передать current_app._get_current_object().

Сигналы и контекст запроса во Flask.

Сигналы Flask полностью поддерживают контекст запроса при получении сигналов. Локальные контекстные переменные постоянно доступны между сигналами flask.request_started() и flask.request_finished, поэтому можно надеяться на глобальный объект flask.g и другое по мере необходимости.
Если flask.signals.signals_available истинно, то будут отправляться следующие сигналы:
  • сигнал flask.request_started отправляется до вызова @app.before_request().
  • сигнал flask.request_finished отправляется после вызова @app.after_request().
  • сигнал flask.got_request_exception отправляется, когда начинается обработка исключения, но до поиска или вызова @app.errorhandler().
  • сигнал flask.request_tearing_down отправляется после вызова @app.teardown_request().

Подписки на сигналы на основе декоратора.

В стороннем модуле blinker с версии 1.1 также можно легко подписаться на сигналы с помощью нового декоратора connect_via():
from flask import template_rendered

@template_rendered.connect_via(app)
def when_template_rendered(sender, template, context, **extra):
    print(f'Template {template.name} is rendered with {context}')

Список всех встроенных сигналов Flask.

Внимание! Для использования представленного ниже API сигналов, необходимо установить сторонний модуль blinker.

flask.signals.signals_available:

Если атрибут flask.signals.signals_available имеет значение True, то система сигналов Flask доступна. Это тот случай, когда установлен сторонний модуль blinker.
>>> from flask import signals
>>> signals.signals_available
# False

Во Flask доступны следующие сигналы:

flask.template_rendered:

Сигнал flask.template_rendered отправляется, когда шаблон был успешно визуализирован. Сигнал вызывается с экземпляром шаблона в качестве template и контекстом в качестве словаря (с именем context).
Пример подписки:
def log_template_renders(sender, template, context, **extra):
    sender.logger.debug('Rendering template "%s" with context %s',
                        template.name or 'string template',
                        context)

from flask import template_rendered
template_rendered.connect(log_template_renders, app)

flask.before_render_template:

Сигнал flask.before_render_template отправляется до процесса рендеринга шаблона. Сигнал вызывается с экземпляром шаблона в качестве template и контекстом в качестве словаря (с именем context).
Пример подписки:
def log_template_renders(sender, template, context, **extra):
    sender.logger.debug('Rendering template "%s" with context %s',
                        template.name or 'string template',
                        context)

from flask import before_render_template
before_render_template.connect(log_template_renders, app)

flask.request_started:

Сигнал flask.request_started отправляется при настройке контекста запроса, до того, как произойдет какая-либо обработка запроса. Так как контекст запроса уже привязан, то подписчик может получить доступ к запросу с помощью стандартных глобальных прокси-объектов, таких как запрос flask.request.
Пример подписки:
def log_request(sender, **extra):
    sender.logger.debug('Request context is set up')

from flask import request_started
request_started.connect(log_request, app)

flask.request_finished:

Сигнал flask.request_finished отправляется прямо перед отправкой ответа клиенту. Ему передается ответ response, который должен быть отправлен именованный ответ.
Пример подписки:
def log_response(sender, response, **extra):
    sender.logger.debug('Request context is about to close down.  '
                        'Response: %s', response)

from flask import request_finished
request_finished.connect(log_response, app)

flask.got_request_exception:

Сигнал flask.got_request_exception отправляется, когда во время обработки запроса возникает необработанное исключение, в том числе при отладке. Исключение передается подписчику как исключение.
Этот сигнал не отправляется для HTTPException или других исключений, для которых зарегистрированы обработчики ошибок, если только исключение не было вызвано этим обработчиком ошибок.
В этом примере показано, как выполнить дополнительное ведение журнала, если возникло теоретическое исключение SecurityException:
from flask import got_request_exception

def log_security_exception(sender, exception, **extra):
    if not isinstance(exception, SecurityException):
        return

    security_logger.exception(
        f"SecurityException at {request.url!r}",
        exc_info=exception,
    )

got_request_exception.connect(log_security_exception, app)

flask.request_tearing_down:

Сигнал flask.request_tearing_down отправляется, когда запрос прерывается. Он вызывается всегда, даже если возникает исключение. В настоящее время функции, прослушивающие этот сигнал, вызываются после обычных обработчиков освобождения ресурсов teardown, но на это нельзя положиться.
Пример подписки:
def close_db_connection(sender, **extra):
    session.close()

from flask import request_tearing_down
request_tearing_down.connect(close_db_connection, app)
Начиная с Flask 0.9, ему также будет передан ключевой аргумент exc, который имеет ссылку на исключение, вызвавшее разрыв, если оно было.

flask.appcontext_tearing_down:

Сигнал flask.appcontext_tearing_down отправляется, когда происходит разрыв контекста приложения. Он вызывается всегда, даже если возникает исключение. В настоящее время функции, прослушивающие этот сигнал, вызываются после обычных обработчиков освобождения ресурсов teardown, но на это нельзя положиться.
Пример подписки:
def close_db_connection(sender, **extra):
    session.close()

from flask import appcontext_tearing_down
appcontext_tearing_down.connect(close_db_connection, app)
Начиная с Flask 0.9, ему также будет передан ключевой аргумент exc, который имеет ссылку на исключение, вызвавшее разрыв, если оно было.

flask.appcontext_pushed:

Сигнал flask.appcontext_pushed отправляется при проталкивании контекста приложения. Отправитель - это приложение app. Сигнал полезен для получения временной информации в модульных тестах. Например, его можно использовать для ранней установки ресурса в глобальный объект flask.g.
Пример использования:
from contextlib import contextmanager
from flask import appcontext_pushed

@contextmanager
def user_set(app, user):
    def handler(sender, **kwargs):
        g.user = user
    with appcontext_pushed.connected_to(handler, app):
        yield
И в тестовом коде:
def test_user_me(self):
    with user_set(app, 'john'):
        c = app.test_client()
        resp = c.get('/users/me')
        assert resp.data == 'username=john'

flask.appcontext_popped:

Сигнал flask.appcontext_popped отправляется при выталкивании контекста приложения. Отправитель - это приложение app. Обычно это соответствует сигналу flask.appcontext_tearing_down.

flask.message_flashed:

Сигнал flask.message_flashed отправляется, когда приложение высвечивает сообщение. Сообщения отправляются как ключевой аргумент message, а категория сообщения - как category.
Пример подписки:
recorded = []
def record(sender, message, category, **extra):
    recorded.append((message, category))

from flask import message_flashed
message_flashed.connect(record, app)

flask.signals.Namespace:

Это псевдоним для blinker.base.Namespace, если модуль blinker доступен, в противном случае - фиктивный класс, который создает ложные сигналы. Этот класс доступен для расширений Flask, которые хотят предоставить ту же резервную систему, что и сам Flask.
  • метод .signal(name, doc=None) создает новый сигнал для этого пространства имен, если модуль blinker доступен, в противном случае возвращает поддельный сигнал, метод отправки которого ничего не сделает, но завершится ошибкой с RuntimeError для всех других операций, включая соединение.