zoukankan      html  css  js  c++  java
  • flask-信号

    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)
    request_started
    同上
    request_finished
    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
    before_render_template
    同上
    template_rendered
    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)
    got_request_exception
    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)
    request_tearing_down
    同上
    appcontext_tearing_down
    同上
    appcontext_tearing_down
    同上
    appcontext_pushed
    同上
    appcontext_popped
    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)
    message_flashed

    自定义信号:

    #!/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__)
     
     
    # 自定义信号
    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()
    View Code

    信号有哪些用途:

    发送邮件:当xx时,发送一封邮件

    写入日志:

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

    import pymysql
    import time
    
    from flask import Flask,Blueprint,request,session,render_template,redirect
    from .forms import LoginForm,RegisterForm
    from  .dbpool import POOL
    from 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('
    ')
    #将函数注册到信号中
    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)
    View Code

     

  • 相关阅读:
    mouse_event模拟鼠标滚轮
    润乾报表配置技术路线
    建筑 物件 开心背单词 读句子,单词,字母,看图例, 翻译,看动画
    文字过渡动画,曲线过渡动画,,使用这个插件assign shape keys
    运动锻炼 开心背单词 读句子,单词,字母,看图例, 翻译,看动画,学英语,轻松背单词,简单背单词
    blender293 内置插件 精度绘画控件,PDT学习003,pdt tangents 切线
    日常用品 背单词 读句子 看图片 读单词 读字母 翻译, 看动画 学英语
    blender293 内置插件 精度绘画控件,PDT学习 precision drawing tools
    乔布斯 背单词 02 读句子 单词 字母 翻译,看动画 学英语 名言 我菜顾我在,我菜故我在,blender加python
    狐狸 和 乌鸦 英语 朗读句子 背单词
  • 原文地址:https://www.cnblogs.com/ctztake/p/8262143.html
Copyright © 2011-2022 走看看