flask-信号,

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在其中添加数据时,自动触发

源码示例

图片 1

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
图片 2

同上

request_finished
图片 3

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
图片 4

同上

template_rendered
图片 5

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
图片 6

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
图片 7

同上

appcontext_tearing_down
图片 8

同上

appcontext_tearing_down
图片 9

同上

appcontext_pushed
图片 10

同上

appcontext_popped
图片 11

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

自定义功率信号:

图片 12

#!/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时,发送意气风发封邮件

写入日志:

亲自去做:当客户登陆成功时,写入一条日志

图片 13

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('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)

View Code

 

 

Flask框架中的功率信号基于blinker,其重大正是让开荒者不过在flask诉求进程中定制一些客商作为
pip3 install blinker 1. 放手时域信号 request…

相关文章

发表评论