Flask框架中的信號(hào)基于blinker，其主要就是讓開發(fā)者可是在flask請(qǐng)求過程中定制一些用戶行為。

pip3 install blinker

使用方法

from flask.signals import request_started

def func1(*args):
    print(111, *args)

def func2(*args):
    print(222, *args)

request_started.connect(func1)  # 將函數(shù)注冊(cè)到信號(hào)中
request_started.connect(func2)
request_started.send()  # 觸發(fā)這個(gè)信號(hào)，執(zhí)行列表中的所有函數(shù)

內(nèi)置信號(hào)

request_started = _signals.signal('request-started')                # 請(qǐng)求到來前執(zhí)行
request_finished = _signals.signal('request-finished')              # 請(qǐng)求結(jié)束后執(zhí)行
 
before_render_template = _signals.signal('before-render-template')  # 模板渲染前執(zhí)行
template_rendered = _signals.signal('template-rendered')            # 模板渲染后執(zhí)行
 
got_request_exception = _signals.signal('got-request-exception')    # 請(qǐng)求執(zhí)行出現(xiàn)異常時(shí)執(zhí)行
 
request_tearing_down = _signals.signal('request-tearing-down')      # 請(qǐng)求執(zhí)行完畢后自動(dòng)執(zhí)行（無論成功與否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 請(qǐng)求上下文執(zhí)行完畢后自動(dòng)執(zhí)行（無論成功與否）
 
appcontext_pushed = _signals.signal('appcontext-pushed')            # 請(qǐng)求上下文push時(shí)執(zhí)行
appcontext_popped = _signals.signal('appcontext-popped')            # 請(qǐng)求上下文pop時(shí)執(zhí)行
message_flashed = _signals.signal('message-flashed')                # 調(diào)用flask在其中添加數(shù)據(jù)時(shí)，自動(dòng)觸發(fā)

源碼示例

appcontext_pushed：請(qǐng)求上下文push時(shí)觸發(fā)

wsgi_app -- ctx.push() -- app_ctx.push()
class AppContext(object):
    def push(self):
        """Binds the app context to the current context."""
        self._refcnt += 1
        if hasattr(sys, 'exc_clear'):
            sys.exc_clear()
        _app_ctx_stack.push(self)
        # ############## 觸發(fā) appcontext_pushed 信號(hào) ##############
        appcontext_pushed.send(self.app)

request_started：在 @before_request之前

wsgi_app -- full_dispatch_request
class Flask(_PackageBoundObject):
    def full_dispatch_request(self):
        self.try_trigger_before_first_request_functions()
        try:
            # ############### 觸發(fā)request_started 信號(hào) ###############
            request_started.send(self)
            rv = self.preprocess_request()
            if rv is None:
                rv = self.dispatch_request()
        except Exception as e:
            rv = self.handle_user_exception(e)
        return self.finalize_request(rv)

request_finished：在@after_request之后

wsgi_app -- full_dispatch_request -- finalize_request
class Flask(_PackageBoundObject):
    def finalize_request(self, rv, from_error_handler=False):
        response = self.make_response(rv)
        try:
            response = self.process_response(response)
            # ############### request_finished 信號(hào) ###############
            request_finished.send(self, response=response)
        except Exception:
            if not from_error_handler:
                raise
            self.logger.exception('Request finalizing failed with an '
                                  'error while handling an error')
        return response

before_render_template 和 template_rendered：在模板渲染時(shí)觸發(fā)

render_template -- _render
def render_template(template_name_or_list, **context):
    ctx = _app_ctx_stack.top
    ctx.app.update_template_context(context)
    return _render(ctx.app.jinja_env.get_or_select_template(template_name_or_list),
                   context, ctx.app)

def _render(template, context, app):
    # ############### before_render_template 信號(hào) ###############
    before_render_template.send(app, template=template, context=context)
    rv = template.render(context)
    
    # ############### template_rendered 信號(hào) ###############
    template_rendered.send(app, template=template, context=context)
    return rv

got_request_exception ：在視圖函數(shù)前后的過程中遇到異常時(shí)觸發(fā)

wsgi_app -- handle_exception
class Flask(_PackageBoundObject):
    def handle_exception(self, e):
        exc_type, exc_value, tb = sys.exc_info()
        # ############### 觸發(fā)got_request_exception 信號(hào) ###############
        got_request_exception.send(self, exception=e)
        handler = self._find_error_handler(InternalServerError())

        if self.propagate_exceptions:
            if exc_value is e:
                reraise(exc_type, exc_value, tb)
            else:
                raise e

        self.log_exception((exc_type, exc_value, tb))
        if handler is None:
            return InternalServerError()
        return self.finalize_request(handler(e), from_error_handler=True)

request_tearing_down 和 appcontext_tearing_down：在wsgi_app最后觸發(fā)

wsgi_app -- ctx.auto_pop() -- ctx.pop() --  app.do_teardown_request() -- app_ctx.pop(exc)

class RequestContext(object):
    def pop(self, exc=_sentinel):
        app_ctx = self._implicit_app_ctx_stack.pop()

        try:
            clear_request = False
            if not self._implicit_app_ctx_stack:
                self.preserved = False
                self._preserved_exc = None
                if exc is _sentinel:
                    exc = sys.exc_info()[1]
                # ################## 觸發(fā) request_tearing_down 信號(hào) ##################
                self.app.do_teardown_request(exc)

                if hasattr(sys, 'exc_clear'):
                    sys.exc_clear()

                request_close = getattr(self.request, 'close', None)
                if request_close is not None:
                    request_close()
                clear_request = True
        finally:
            rv = _request_ctx_stack.pop()

            # get rid of circular dependencies at the end of the request
            # so that we don't require the GC to be active.
            if clear_request:
                rv.request.environ['werkzeug.request'] = None

            # Get rid of the app as well if necessary.
            if app_ctx is not None:
                # ################## 觸發(fā) appcontext_tearing_down和appcontext_popped##################
                app_ctx.pop(exc)

            assert rv is self, 'Popped wrong request context.  ' \
                '(%r instead of %r)' % (rv, self)

class AppContext(object):
    def pop(self, exc=_sentinel):
        """Pops the app context."""
        try:
            self._refcnt -= 1
            if self._refcnt <= 0:
                if exc is _sentinel:
                    exc = sys.exc_info()[1]
                # ################## 觸發(fā) appcontext_tearing_down##################
                self.app.do_teardown_appcontext(exc)
        finally:
            rv = _app_ctx_stack.pop()
        assert rv is self, 'Popped wrong app context.  (%r instead of %r)' \
            % (rv, self)
        # ################## 觸發(fā) appcontext_popped##################
        appcontext_popped.send(self.app)

class Flask(_PackageBoundObject):
    def do_teardown_request(self, exc=_sentinel):
        if exc is _sentinel:
            exc = sys.exc_info()[1]
        funcs = reversed(self.teardown_request_funcs.get(None, ()))
        bp = _request_ctx_stack.top.request.blueprint
        if bp is not None and bp in self.teardown_request_funcs:
            funcs = chain(funcs, reversed(self.teardown_request_funcs[bp]))
        for func in funcs:
            func(exc)
        # ################## 觸發(fā) request_tearing_down 信號(hào) ##################
        request_tearing_down.send(self, exc=exc)

    def do_teardown_appcontext(self, exc=_sentinel):
        if exc is _sentinel:
            exc = sys.exc_info()[1]
        for func in reversed(self.teardown_appcontext_funcs):
            func(exc)
        # ################## 觸發(fā) appcontext_tearing_down信號(hào) ##################
        appcontext_tearing_down.send(self, exc=exc)

message_flashed：在調(diào)用flash函數(shù)時(shí)觸發(fā)

def flash(message, category='message'):
    flashes = session.get('_flashes', [])
    flashes.append((category, message))
    session['_flashes'] = flashes
    # ############### 觸發(fā) message_flashed 信號(hào) ###############
    message_flashed.send(current_app._get_current_object(),
                         message=message, category=category)

自定義信號(hào)

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Flask, current_app, flash, render_template
from flask.signals import _signals
 
app = Flask(import_name=__name__)
 
 
# 自定義信號(hào)
xxxxx = _signals.signal('xxxxx')
 
 
def func(sender, *args, **kwargs):
    print(sender)
 
# 自定義信號(hào)中注冊(cè)函數(shù)
xxxxx.connect(func)
 
 
@app.route("/x")
def index():
    # 觸發(fā)信號(hào)
    xxxxx.send('123123', k1='v1')
    return 'Index'
 
 
if __name__ == '__main__':
    app.run()