Сигналы

Добавлено в версии 0.6.

В Flask, начиная с версии 0.6, есть интегрированная поддержка для передачи сигналов. Эта поддержка обеспечивается великолепной библиотекой blinker, при отсутствии которой она элегантно отключается.

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

Внутри Flask уже есть пара сигналов, а другие расширения могут давать дополнительные. Также имейте ввиду, что сигналы предназначены для уведомления подписчиков и не должны им содействовать в изменении данных. Вы заметите, что есть такие сигналы, которые появляются для того, чтобы сделать то же самое, что делают некоторые из встроенных декораторов (например: request_started очень похож на before_request()). Однако, есть разница в том, как они работают. Например, обработчик before_request(), который находится в ядре, выполняется в соответствии с определённым порядком и способен с помощью возврата ответа прервать запрос на ранней стадии. Напротив, порядок выполнения для всех обработчиков сигналов, которые в процессе работы не изменяют какие-либо данные, не определён.

Большим преимуществом сигналов перед обработчиками является то, что вы можете в один миг спокойно на них подписаться. Такие временные подписки могут оказаться полезными, например, при тестировании модулей. Например, сигналы прекрасно подходят для того, чтобы узнать, что было отрисовано шаблоном в рамках запроса.

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

Чтобы подписаться на сигнал, вы можете использовать его метод connect(). Первый аргумент - это функция, которая должна быть вызвана в момент подачи сигнала, второй необязательный аргумент определяет отправителя. Для того, чтобы отписаться от сигнала, вы можете использовать метод disconnect().

Для всех сигналов ядра Flask, отправителем является приложение, которое выдало этот сигнал. Когда вы подписываетесь на сигнал и не хотите слушать сигналы от всех приложений, убедитесь, что указали отправителя. Особенно это верно, если вы взялись за разработку расширения 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

Чтобы ваши вызовы не порушились в ситуации, если Flask вставит для сигнала новые аргументы, убедитесь, что подписываетесь с дополнительным аргументом **extra

Вся отрисовка шаблона внутри кода, вызванного изнутри тела блока with приложения app, теперь будет записана в переменную templates. Всякий раз при отрисовке шаблона, к ней будут добавлены объект шаблона и контекст.

Кроме того, есть удобный метод помощника (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 API

Метод connected_to() появился в Blinker, начиная с версии 1.1.

Создание сигналов

Если вы хотите использовать сигналы в ваших собственных приложениях, вы можете напрямую использовать библиотеку blinker. Самый общий случай - это именованные сигналы в пользовательском Namespace.. Это то, что можно порекомендовать чаще всего:

from blinker import Namespace
my_signals = Namespace()

Теперь вы можете создать такой вот сигнал:

model_saved = my_signals.signal('model-saved')

Имя сигнала здесь делает его уникальным, а также упрощает отладку. Вы можете получить доступ к имени сигнала через атрибут name.

Для продвинутых разработчиков

Если вы пишете расширения для Flask и хотите изящно обойти ситуацию отсутствия blinker, вы можете сделать это с использованием класса flask.signals.Namespace.

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

Если вы хотите послать сигнал, вы можете это сделать, вызвав метод send(). Он принимает отправителя в качестве первого аргумента и опционально - некоторые именованные аргументы, которые будут перенаправлены подписчикам сигнала:

class Model(object):
    ...

    def save(self):
        model_saved.send(self)

Всегда пытайтесь выбирать верного отправителя. Если у вас есть класс, который посылает сигнал, в качестве отправителя передайте self. Если вы посылаете сигнал из случайной функции, в качестве отправителя можно указать current_app._get_current_object().

Передача в качестве отправителей посредников (Proxies)

Никогда не передавайте сигналу в качестве отправителя current_app. Вместо этого используйте current_app._get_current_object(). Это объясняется тем, что current_app - это посредник, а не объект реального приложения.

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

Сигналы (при их получении) полностью поддерживают Контекст запроса Flask. Контекстно-локальные переменные постоянно доступны между request_started и request_finished, так что вы можете при необходимости рассчитывать на flask.g или другие. Заметим, что есть ограничения, описанные в Отправка сигналов и описании сигнала request_tearing_down.

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

С Blinker 1.1 вы также можете легко подписаться на сигналы с помощью нового декоратора connect_via():

from flask import template_rendered

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

Сигналы ядра

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

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.request_started

Сигнал посылается перед запуском обработки любого запроса, но после того, как контекст запроса уже установлен. Поскольку контекст запроса уже привязан, подписчик может получить доступ к запросу через стандартные глобальные прокси-посредники, такие как 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

Этот сигнал посылается прямо перед отправкой ответа клиенту. Он передаётся для отправки ответу с именем 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

Этот сигнал посылается при возникновении исключения во время обработки запроса. Он посылается до того, как умрёт стандартная обработка исключения, и даже в режиме отладки, когда обработки исключений не происходит. Само по себе исключение передаётся подписчику как exception.

Пример подписчика:

def log_exception(sender, exception, **extra):
    sender.logger.debug('Got exception during processing: %s', exception)

from flask import got_request_exception
got_request_exception.connect(log_exception, app)
flask.request_tearing_down

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

Пример подписчика:

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

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

Пример подписчика:

def close_db_connection(sender, **extra):
    session.close()

from flask import appcontext_tearing_down
appcontext_tearing_down.connect(close_db_connection, app)

Здесь также будет передан аргумент с именем exc, который ссылается на исключение, которое вызвало демонтаж, если таковое было.

flask.appcontext_pushed

Этот сигнал посылается, при запуске контекста приложения. Отправитель - приложение. Обычно он может быть полезен для юнит-тестов, когда нужно временно выцепить информацию. Например, он может быть использован, чтобы на ранней стадии установить значение ресурса для объекта 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'

Добавлено в версии 0.10.

flask.appcontext_popped

Этот сигнал посылается при сбросе контекста приложения. Отправитель - приложение. Обычно это происходит вместе с появлением сигнала appcontext_tearing_down.

Добавлено в версии 0.10.

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)

Добавлено в версии 0.10.

Оригинал этой страницы