zoukankan      html  css  js  c++  java
  • Flask框架 (四)—— 请求上下文源码分析、g对象、第三方插件(flask_session、flask_script、wtforms)、信号

    请求上下文源码分析、g对象、第三方插件(flask_session、flask_script、wtforms)、信号

    一、请求上下文源码分析

    第一阶段:将ctx(request,session)放到Local对象上
    				   
    第二阶段:视图函数导入:request/session 
    request.method
    	-LocalProxy对象.method,执行getattr方法,getattr(self._get_current_object(), name)
    		-self._get_current_object()返回return self.__local(),self.__local(),在LocakProxy实例化的时候,object.__setattr__(self, '_LocalProxy__local', local),此处local就是:partial(_lookup_req_object, 'request')
    
    	-def _lookup_req_object(name):
    			top = _request_ctx_stack.top #_request_ctx_stack 就是LocalStack()对象,top方法把ctx取出来
    			if top is None:
    				raise RuntimeError(_request_ctx_err_msg)
    			return getattr(top, name)#获取ctx中的request或session对象
    
    第三阶段:请求处理完毕
    		- 获取session并保存到cookie
    		- 将ctx删除
    

    程序运行,两个LocalStack()对象,一个里面放request和session,另一个放g和current_app

    二、g对象

    1、什么是g对象

    专门用来存储用户信息的g对象,g的全称的为global

    g对象在一次请求中的所有的代码的地方,都是可以使用的

    2、g对象和session的区别

    session对象是可以跨request的,只要session还未失效,不同的request的请求会获取到同一个session
    但是g对象不是,g对象不需要管过期时间,请求一次就g对象就改变了一次,或者重新赋值了一次
    

    三、flask-session

    1、作用

    将默认保存的签名cookie中的值,保存到 redis/memcached/file/Mongodb/SQLAlchemy

    2、安装

    pip3 install flask-session
    

    3、使用

    (1)使用redis保存session

    from flask import Flask,session
    from flask_session import RedisSessionInterface
    import redis
    app = Flask(__name__)
    conn=redis.Redis(host='127.0.0.1',port=6379)
    #use_signer是否对key签名
    app.session_interface=RedisSessionInterface(conn,key_prefix='lqz')
    @app.route('/')
    def hello_world():
        session['name']='lqz'
        return 'Hello World!'
    
    if __name__ == '__main__':
        app.run()
    

    (2)通过修改配置实现可插拔式保存session

    from redis import Redis
    from flask.ext.session import Session
    app.config['SESSION_TYPE'] = 'redis'   # 指定session保存的方式为保存在redis中 
    app.config['SESSION_REDIS'] = Redis(host='192.168.0.94',port='6379')
    Session(app)
    

    问题:设置cookie时,如何设定关闭浏览器则cookie失效。

    response.set_cookie('k','v',exipre=None)	# 这样设置即可
    # 在session中设置
    app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',permanent=False)
    # 一般不用,我们一般都设置超时时间,多长时间后失效
    

    问题:cookie默认超时时间是多少?如何设置超时时间

    # 源码expires = self.get_expiration_time(app, session)
    'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),  # 这个配置文件控制
    

    四、flask-script

    1、什么是flask_script

    用于实现类似于django中 python3 manage.py runserver ...类似的命令

    2、安装

    pip3 install flask-script
    

    3、使用

    from flask_script import Manager
    app = Flask(__name__)
    manager=Manager(app)
    ...
    if __name__ == '__main__':
        manager.run()
    #以后在执行,直接:python3 manage.py runserver
    #python3 manage.py runserver --help
    

    4、自定义命令

    @manager.command
    def custom(arg):
        """
        自定义命令
        python manage.py custom 123
        :param arg:
        :return:
        """
        print(arg)
    
    
    @manager.option('-n', '--name', dest='name')
    #@manager.option('-u', '--url', dest='url')
    def cmd(name, url):
        """
        自定义命令(-n也可以写成--name)
        执行: python manage.py  cmd -n lqz -u http://www.oldboyedu.com
        执行: python manage.py  cmd --name lqz --url http://www.oldboyedu.com
        :param name:
        :param url:
        :return:
        """
        print(name, url)
    # 作用;
    # 把excel的数据导入数据库,定制个命令,去执行
    

    五、数据库连接池

    1、pymsql链接数据库

    import pymysql
    
    conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123456', db='s8day127db')
    cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
    # cursor.execute("select id,name from users where name=%s and pwd=%s",['lqz','123',])
    cursor.execute("select id,name from users where name=%(user)s and pwd=%(pwd)s",{'user':'lqz','pwd':'123'})
    obj = cursor.fetchone()
    conn.commit()
    cursor.close()
    conn.close()
    
    print(obj)
    

    2、数据库连接池版

    setting.py

    from datetime import timedelta
    from redis import Redis
    import pymysql
    from DBUtils.PooledDB import PooledDB, SharedDBConnection
    
    class Config(object):
        DEBUG = True
        SECRET_KEY = "umsuldfsdflskjdf"
        PERMANENT_SESSION_LIFETIME = timedelta(minutes=20)
        SESSION_REFRESH_EACH_REQUEST= True
        SESSION_TYPE = "redis"
        PYMYSQL_POOL = PooledDB(
            creator=pymysql,  # 使用链接数据库的模块
            maxconnections=6,  # 连接池允许的最大连接数,0和None表示不限制连接数
            mincached=2,  # 初始化时,链接池中至少创建的空闲的链接,0表示不创建
            maxcached=5,  # 链接池中最多闲置的链接,0和None不限制
            maxshared=3,
            # 链接池中最多共享的链接数量,0和None表示全部共享。PS: 无用,因为pymysql和MySQLdb等模块的 threadsafety都为1,所有值无论设置为多少,_maxcached永远为0,所以永远是所有链接都共享。
            blocking=True,  # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错
            maxusage=None,  # 一个链接最多被重复使用的次数,None表示无限制
            setsession=[],  # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."]
            ping=0,
            # ping MySQL服务端,检查是否服务可用。# 如:0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
            host='127.0.0.1',
            port=3306,
            user='root',
            password='123456',
            database='s8day127db',
            charset='utf8'
        )
    
    class ProductionConfig(Config):
        SESSION_REDIS = Redis(host='192.168.0.94', port='6379')
    
    
    
    class DevelopmentConfig(Config):
        SESSION_REDIS = Redis(host='127.0.0.1', port='6379')
    
    
    class TestingConfig(Config):
        pass
    

    utils/sql.py

    import pymysql
    from settings import Config
    class SQLHelper(object):
    
        @staticmethod
        def open(cursor):
            POOL = Config.PYMYSQL_POOL
            conn = POOL.connection()
            cursor = conn.cursor(cursor=cursor)
            return conn,cursor
    
        @staticmethod
        def close(conn,cursor):
            conn.commit()
            cursor.close()
            conn.close()
    
        @classmethod
        def fetch_one(cls,sql,args,cursor =pymysql.cursors.DictCursor):
            conn,cursor = cls.open(cursor)
            cursor.execute(sql, args)
            obj = cursor.fetchone()
            cls.close(conn,cursor)
            return obj
    
        @classmethod
        def fetch_all(cls,sql, args,cursor =pymysql.cursors.DictCursor):
            conn, cursor = cls.open(cursor)
            cursor.execute(sql, args)
            obj = cursor.fetchall()
            cls.close(conn, cursor)
            return obj
    

    使用:

    obj = SQLHelper.fetch_one("select id,name from users where name=%(user)s and pwd=%(pwd)s", form.data)
    

    六、wtforms

    1、安装

    pip3 install wtforms
    

    2、使用

    (1)使用一

    from flask import Flask, render_template, request, redirect
    from wtforms import Form
    from wtforms.fields import simple
    from wtforms import validators
    from wtforms import widgets
    
    app = Flask(__name__, template_folder='templates')
    
    app.debug = True
    
    
    class LoginForm(Form):
        # 字段(内部包含正则表达式)
        name = simple.StringField(
            label='用户名',
            validators=[
                validators.DataRequired(message='用户名不能为空.'),
                validators.Length(min=6, max=18, message='用户名长度必须大于%(min)d且小于%(max)d')
            ],
            widget=widgets.TextInput(), # 页面上显示的插件
            render_kw={'class': 'form-control'}
    
        )
        # 字段(内部包含正则表达式)
        pwd = simple.PasswordField(
            label='密码',
            validators=[
                validators.DataRequired(message='密码不能为空.'),
                validators.Length(min=8, message='用户名长度必须大于%(min)d'),
                validators.Regexp(regex="^(?=.*[a-z])(?=.*[A-Z])(?=.*d)(?=.*[$@$!%*?&])[A-Za-zd$@$!%*?&]{8,}",
                                  message='密码至少8个字符,至少1个大写字母,1个小写字母,1个数字和1个特殊字符')
    
            ],
            widget=widgets.PasswordInput(),
            render_kw={'class': 'form-control'}
        )
    
    
    
    @app.route('/login', methods=['GET', 'POST'])
    def login():
        if request.method == 'GET':
            form = LoginForm()
            return render_template('login.html', form=form)
        else:
            form = LoginForm(formdata=request.form)
            if form.validate():
                print('用户提交数据通过格式验证,提交的值为:', form.data)
            else:
                print(form.errors)
            return render_template('login.html', form=form)
    
    if __name__ == '__main__':
        app.run()
    

    login.html

    <!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>Title</title>
    </head>
    <body>
    <h1>登录</h1>
    <form method="post">
        <p>{{form.name.label}} {{form.name}} {{form.name.errors[0] }}</p>
    
        <p>{{form.pwd.label}} {{form.pwd}} {{form.pwd.errors[0] }}</p>
        <input type="submit" value="提交">
    </form>
    </body>
    </html>
    
    

    (2)使用2

    from flask import Flask, render_template, request, redirect
    from wtforms import Form
    from wtforms.fields import core
    from wtforms.fields import html5
    from wtforms.fields import simple
    from wtforms import validators
    from wtforms import widgets
    
    app = Flask(__name__, template_folder='templates')
    app.debug = True
    
    
    
    class RegisterForm(Form):
        name = simple.StringField(
            label='用户名',
            validators=[
                validators.DataRequired()
            ],
            widget=widgets.TextInput(),
            render_kw={'class': 'form-control'},
            default='alex'
        )
    
        pwd = simple.PasswordField(
            label='密码',
            validators=[
                validators.DataRequired(message='密码不能为空.')
            ],
            widget=widgets.PasswordInput(),
            render_kw={'class': 'form-control'}
        )
    
        pwd_confirm = simple.PasswordField(
            label='重复密码',
            validators=[
                validators.DataRequired(message='重复密码不能为空.'),
                validators.EqualTo('pwd', message="两次密码输入不一致")
            ],
            widget=widgets.PasswordInput(),
            render_kw={'class': 'form-control'}
        )
    
        email = html5.EmailField(
            label='邮箱',
            validators=[
                validators.DataRequired(message='邮箱不能为空.'),
                validators.Email(message='邮箱格式错误')
            ],
            widget=widgets.TextInput(input_type='email'),
            render_kw={'class': 'form-control'}
        )
    
        gender = core.RadioField(
            label='性别',
            choices=(
                (1, '男'),
                (2, '女'),
            ),
            coerce=int # “1” “2”
         )
        city = core.SelectField(
            label='城市',
            choices=(
                ('bj', '北京'),
                ('sh', '上海'),
            )
        )
    
        hobby = core.SelectMultipleField(
            label='爱好',
            choices=(
                (1, '篮球'),
                (2, '足球'),
            ),
            coerce=int
        )
    
        favor = core.SelectMultipleField(
            label='喜好',
            choices=(
                (1, '篮球'),
                (2, '足球'),
            ),
            widget=widgets.ListWidget(prefix_label=False),
            option_widget=widgets.CheckboxInput(),
            coerce=int,
            default=[1, 2]
        )
    
        def __init__(self, *args, **kwargs):
            super(RegisterForm, self).__init__(*args, **kwargs)
            self.favor.choices = ((1, '篮球'), (2, '足球'), (3, '羽毛球'))
    
        def validate_pwd_confirm(self, field):
            """
            自定义pwd_confirm字段规则,例:与pwd字段是否一致
            :param field:
            :return:
            """
            # 最开始初始化时,self.data中已经有所有的值
    
            if field.data != self.data['pwd']:
                # raise validators.ValidationError("密码不一致") # 继续后续验证
                raise validators.StopValidation("密码不一致")  # 不再继续后续验证
    
    
    @app.route('/register', methods=['GET', 'POST'])
    def register():
        if request.method == 'GET':
            form = RegisterForm(data={'gender': 2,'hobby':[1,]}) # initial
            return render_template('register.html', form=form)
        else:
            form = RegisterForm(formdata=request.form)
            if form.validate():
                print('用户提交数据通过格式验证,提交的值为:', form.data)
            else:
                print(form.errors)
            return render_template('register.html', form=form)
    
    
    
    if __name__ == '__main__':
        app.run()
    
    
    <!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>Title</title>
    </head>
    <body>
    <h1>用户注册</h1>
    <form method="post" novalidate style="padding:0  50px">
        {% for field in form %}
        <p>{{field.label}}: {{field}} {{field.errors[0] }}</p>
        {% endfor %}
        <input type="submit" value="提交">
    </form>
    </body>
    </html>
    
    

    七、信号

    1、什么是信号

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

    2、安装

    pip3 install blinker
    
    

    3、内置信号

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

    4、使用信号

    from flask import Flask,signals,render_template
    
    app = Flask(__name__)
    
    # 往信号中注册函数
    def func(*args,**kwargs):
        print('触发型号',args,kwargs)
    signals.request_started.connect(func)
    
    # 触发信号: signals.request_started.send()
    @app.before_first_request
    def before_first1(*args,**kwargs):
        pass
    @app.before_first_request
    def before_first2(*args,**kwargs):
        pass
    
    @app.before_request
    def before_first3(*args,**kwargs):
        pass
    
    @app.route('/',methods=['GET',"POST"])
    def index():
        print('视图')
        return render_template('index.html')
    
    
    if __name__ == '__main__':
        app.wsgi_app
        app.run()
    
    

    5、一个流程中的信号触发点

    a. before_first_request
    b. 触发 request_started 信号
    c. before_request
    d. 模板渲染
    	渲染前的信号 before_render_template.send(app, template=template, context=context)
    		rv = template.render(context) # 模板渲染
    	渲染后的信号 template_rendered.send(app, template=template, context=context)
    e. after_request
    f. session.save_session()
    g. 触发 request_finished信号		
    	如果上述过程出错:
    		触发错误处理信号 got_request_exception.send(self, exception=e)
    			
    h. 触发信号 request_tearing_down
    
    

    6、自定义信号

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

    八、多app应用(使用蓝图即可实现)

    from werkzeug.wsgi import DispatcherMiddleware
    from werkzeug.serving import run_simple
    from flask import Flask, current_app
    app1 = Flask('app01')
    app2 = Flask('app02')
    
    @app1.route('/index')
    def index():
        return "app01"
    
    @app2.route('/index2')
    def index2():
        return "app2"
    
    # http://www.oldboyedu.com/index
    # http://www.oldboyedu.com/sec/index2
    dm = DispatcherMiddleware(app1, {
        '/sec': app2,
    })
    
    if __name__ == "__main__":
        run_simple('localhost', 5000, dm)
    
    
    

  • 相关阅读:
    【C#进阶系列】06 类型和成员基础
    纪中5日T1 1564. 旅游
    纪中17日T1 2321. 方程
    纪中17日T2 2322. capacitor
    纪中10日T1 2313. 动态仙人掌
    纪中14日听课小结 图论 最短路 二分图 差分约束
    一个抓猫的游戏 消遣GAME 持续更新中!
    洛谷P1464 Function  HDU P1579 Function Run Fun
    洛谷P1976 鸡蛋饼
    纪中12日T1 2307. 选择
  • 原文地址:https://www.cnblogs.com/linagcheng/p/10402468.html
Copyright © 2011-2022 走看看