Сигналы Flask помогают отделить приложения, отправляя уведомления, когда действия происходят в другом месте основной платформы или в других расширениях Flask. Короче говоря, сигналы позволяют определенным отправителям уведомлять подписчиков о том, что что-то произошло.
Поддержка сигналов обеспечивается сторонним модулем blinker и будет изящно отменена, если он не установлен.
Flask поставляется с двумя настроенными сигналами, другие расширения могут предоставить больше сигналов. Также имейте в виду, что сигналы предназначены для уведомления подписчиков и не должны побуждать подписчиков изменять данные. Есть сигналы, которые, например делают то же самое, что и некоторые из встроенных декораторов (например: сигнал flask.request_started() очень похож на декоратор @app.before_request()), но есть различия в том, как они работают. Например, основной обработчик @app.before_request() выполняется в определенном порядке и может преждевременно прервать запрос, вернув ответ. Напротив, все сигналы выполняются в неопределенном порядке и не изменяют никаких данных.
Большим преимуществом сигналов перед обработчиками является то, что можно безопасно подписаться на них всего за несколько строк кода. Эти временные подписки полезны, например, для модульного тестирования. Допустим, необходимо знать, какие шаблоны были отрисованы как часть запроса: сигналы позволяют сделать именно это.
Чтобы подписаться на сигнал, необходимо использовать метод сигнала 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]
Если нужно использовать сигналы в собственном приложении, то можно использовать сторонний модуль 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.request_started() и flask.request_finished, поэтому можно надеяться на глобальный объект flask.g и другое по мере необходимости.
Атрибут flask.signals.signals_available устарел с версии 2.3.0. С версии Flask 2.3.0 сигналы доступны всегда:
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().С версии Flask 2.3.0 сигналы поддерживают асинхронные функции.
В стороннем модуле 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}')
Внимание! Для использования представленного ниже API сигналов, необходимо установить сторонний модуль blinker.
flask.signals.signals_available True, если установлен сторонний модуль blinker,flask.template_rendered шаблон был успешно визуализирован,flask.before_render_template отправляется до рендеринга шаблона,flask.request_started отправляется до начала обработки запроса,flask.request_finished отправляется прямо перед отправкой ответа,flask.got_request_exception отправляется при возникновении необработанного исключения,flask.request_tearing_down отправляется, когда запрос прерывается,flask.appcontext_tearing_down отправляется, когда происходит разрыв контекста приложения,flask.appcontext_pushed отправляется при проталкивании контекста приложения,flask.appcontext_popped отправляется при выталкивании контекста приложения,flask.message_flashed отправляется, когда приложение высвечивает сообщение,flask.signals.Namespace псевдоним для blinker.base.Namespace.flask.signals.signals_available:Если атрибут flask.signals.signals_available имеет значение True, то система сигналов Flask доступна. Это тот случай, когда установлен сторонний модуль blinker.
>>> from flask import signals >>> signals.signals_available # False
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 для всех других операций, включая соединение.