Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为

pip3 install blinker

　1. 内置信号

request_started = _signals.signal('request-started')                # 请求到来前执行request_finished = _signals.signal('request-finished')              # 请求结束后执行 before_render_template = _signals.signal('before-render-template')  # 模板渲染前执行template_rendered = _signals.signal('template-rendered')            # 模板渲染后执行 got_request_exception = _signals.signal('got-request-exception')    # 请求执行出现异常时执行 request_tearing_down = _signals.signal('request-tearing-down')      # 请求执行完毕后自动执行（无论成功与否）appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 请求上下文执行完毕后自动执行（无论成功与否） appcontext_pushed = _signals.signal('appcontext-pushed')            # 请求上下文push时执行appcontext_popped = _signals.signal('appcontext-popped')            # 请求上下文pop时执行message_flashed = _signals.signal('message-flashed')                # 调用flask在其中添加数据时，自动触发

源码示例

class Flask(_PackageBoundObject):    def full_dispatch_request(self):               self.try_trigger_before_first_request_functions()        try:            # ############### 触发request_started 信号 ###############            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)        response = self.make_response(rv)        response = self.process_response(response)        # ############### request_finished 信号 ###############        request_finished.send(self, response=response)        return response    def wsgi_app(self, environ, start_response):                ctx = self.request_context(environ)        ctx.push()        error = None        try:            try:                response = self.full_dispatch_request()            except Exception as e:                error = e                response = self.make_response(self.handle_exception(e))            return response(environ, start_response)        finally:            if self.should_ignore_error(error):                error = None            ctx.auto_pop(error)

同上

def render_template(template_name_or_list, **context):    """Renders a template from the template folder with the given    context.    :param template_name_or_list: the name of the template to be                                  rendered, or an iterable with template names                                  the first one existing will be rendered    :param context: the variables that should be available in the                    context of the template.    """    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):    """Renders the template and fires the signal"""    # ############### before_render_template 信号 ###############    before_render_template.send(app, template=template, context=context)    rv = template.render(context)        # ############### template_rendered 信号 ###############    template_rendered.send(app, template=template, context=context)    return rv

同上

class Flask(_PackageBoundObject):    def handle_exception(self, e):               exc_type, exc_value, tb = sys.exc_info()        # ############### got_request_exception 信号 ###############        got_request_exception.send(self, exception=e)        handler = self._find_error_handler(InternalServerError())        if self.propagate_exceptions:            # if we want to repropagate the exception, we can attempt to            # raise it with the whole traceback in case we can do that            # (the function was actually called from the except part)            # otherwise, we just raise the error again            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 handler(e)    def wsgi_app(self, environ, start_response):                ctx = self.request_context(environ)        ctx.push()        error = None        try:            try:                response = self.full_dispatch_request()            except Exception as e:                error = e                # 这里这里这里这里这里这里这里这里这里这里这里这里 #                response = self.make_response(self.handle_exception(e))            return response(environ, start_response)        finally:            if self.should_ignore_error(error):                error = None            ctx.auto_pop(error)

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)        # ############## 触发 appcontext_pushed 信号 ##############        appcontext_pushed.send(self.app)    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]                # ############## 触发 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)        # ############## 触发 appcontext_popped 信号 ##############        appcontext_popped.send(self.app)class RequestContext(object):    def push(self):        top = _request_ctx_stack.top        if top is not None and top.preserved:            top.pop(top._preserved_exc)        app_ctx = _app_ctx_stack.top        if app_ctx is None or app_ctx.app != self.app:                        # ####################################################            app_ctx = self.app.app_context()            app_ctx.push()            self._implicit_app_ctx_stack.append(app_ctx)        else:            self._implicit_app_ctx_stack.append(None)        if hasattr(sys, 'exc_clear'):            sys.exc_clear()        _request_ctx_stack.push(self)        # Open the session at the moment that the request context is        # available. This allows a custom open_session method to use the        # request context (e.g. code that access database information        # stored on `g` instead of the appcontext).        self.session = self.app.open_session(self.request)        if self.session is None:            self.session = self.app.make_null_session()class Flask(_PackageBoundObject):    def wsgi_app(self, environ, start_response):                ctx = self.request_context(environ)        ctx.push()        error = None        try:            try:                response = self.full_dispatch_request()            except Exception as e:                error = e                response = self.make_response(self.handle_exception(e))            return response(environ, start_response)        finally:            if self.should_ignore_error(error):                error = None            ctx.auto_pop(error)    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]                # ################## 触发 request_tearing_down 信号 ##################                self.app.do_teardown_request(exc)                # If this interpreter supports clearing the exception information                # we do that now.  This will only go into effect on Python 2.x,                # on 3.x it disappears automatically at the end of the exception                # stack.                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:                # ####################################################                app_ctx.pop(exc)            assert rv is self, 'Popped wrong request context.  ' \                '(%r instead of %r)' % (rv, self)    def auto_pop(self, exc):        if self.request.environ.get('flask._preserve_context') or \           (exc is not None and self.app.preserve_context_on_exception):            self.preserved = True            self._preserved_exc = exc        else:            self.pop(exc)

同上

同上

同上

同上

def flash(message, category='message'):    """Flashes a message to the next request.  In order to remove the    flashed message from the session and to display it to the user,    the template has to call :func:`get_flashed_messages`.    .. versionchanged:: 0.3       `category` parameter added.    :param message: the message to be flashed.    :param category: the category for the message.  The following values                     are recommended: ``'message'`` for any kind of message,                     ``'error'`` for errors, ``'info'`` for information                     messages and ``'warning'`` for warnings.  However any                     kind of string can be used as category.    """    # Original implementation:    #    #     session.setdefault('_flashes', []).append((category, message))    #    # This assumed that changes made to mutable structures in the session are    # are always in sync with the session object, which is not true for session    # implementations that use external storage for keeping their keys/values.    flashes = session.get('_flashes', [])    flashes.append((category, message))    session['_flashes'] = flashes    # ############### 触发 message_flashed 信号 ###############    message_flashed.send(current_app._get_current_object(),                         message=message, category=category)

自定义信号:

#!/usr/bin/env python# -*- coding:utf-8 -*-from flask import Flask, current_app, flash, render_templatefrom flask.signals import _signals app = Flask(import_name=__name__)  # 自定义信号xxxxx = _signals.signal('xxxxx')  def func(sender, *args, **kwargs):    print(sender) # 自定义信号中注册函数xxxxx.connect(func)  @app.route("/x")def index():    # 触发信号    xxxxx.send('123123', k1='v1')    return 'Index'  if __name__ == '__main__':    app.run()

信号有哪些用途:

发送邮件：当xx时，发送一封邮件

写入日志:

示例:当用户登录成功时，写入一条日志

import pymysqlimport timefrom flask import Flask,Blueprint,request,session,render_template,redirectfrom .forms import LoginForm,RegisterFormfrom  .dbpool import POOLfrom flask.signals import _signals#自定义信号login_log=_signals.signal('login_log')user=Blueprint('user',__name__,)#登录写入日志def writeLog(*args,**kwargs):    with open('log.txt','a',encoding='gbk')as f:        current_time=time.strftime('%Y-%m-%d:%H:%M:%S',time.localtime())        message='%s于%s登录'%(session.get('user').get('username'),current_time)        f.write(message)        f.write('\n')#将函数注册到信号中login_log.connect(writeLog)@user.route('/login',methods=['GET','POST'])def login():    if request.method=='GET':        form=LoginForm()        print(form)        return render_template('login.html',form=form)    else:        form=LoginForm(request.form)        if form.validate():            #数据库连接池创建链接            conn=POOL.connection()            cursor=conn.cursor(cursor=pymysql.cursors.DictCursor)            sql='select * from userinfo where username=%s and pwd=%s '            print(form.data)            username=form.data.get('name')            pwd=form.data.get('pwd')            cursor.execute(sql,[username,pwd])            user=cursor.fetchone()            cursor.close()            conn.close()            if user:                session['user']=user                #写入日志                login_log.send()                return render_template('index.html',user=user)            return render_template('login.html',form=form,msg='用户名或密码错误')        return render_template('login.html',form=form,)@user.route('/regist',methods=['GET','POST'])def register():    if request.method=='GET':        form=RegisterForm()        return render_template('register.html',form=form)    else:        form=RegisterForm(request.form)        if form.validate():            print(form.data)            name=form.data.get('name')            pwd=form.data.get('pwd')            email=form.data.get('email')            gender=form.data.get('gender')            city=form.data.get('city')            hobbys=form.data.get('hobby')            conn=POOL.connection()            cursor=conn.cursor(cursor=pymysql.cursors.DictCursor)            sql="insert into userinfo(username,pwd,email,gender,city)VALUES(%s,%s,%s,%s,%s)"            cursor.execute(sql,[name,pwd,email,gender,city])            conn.commit()            cursor.close()            conn.close()            return redirect('/login')        return render_template('register.html', form=form)