Flaskフレームワーク(4)——要求コンテキストソース分析、gオブジェクト、サードパーティプラグイン(flask_session、flask_script、wtforms)、信号

Flaskフレームワーク(4)——要求コンテキストソース分析、gオブジェクト、サードパーティプラグイン(flask_session、flask_script、wtforms)、信号
目次

要求コンテキストソース分析、gオブジェクト、サードパーティプラグイン(flask_session、flask_script、wtforms)、信号

一、要求コンテキストソース分析

二、g対象

1、gオブジェクトとは何か

2、gオブジェクトとセッションの違い

三、flask-session

1、作用

2、取付

3、使用

四、flask-script

1、flaskとは何かscript

2、取付

3、使用

4、カスタムコマンド

五、データベース接続プール

1、pymsqlリンクデータベース

2、データベース接続プール版

六、wtforms

1、インストール

2、使用

七、信号

1、信号とは何か

2、取付

3、内蔵信号

4、使用信号

5、1つのフローにおける信号トリガ

6、カスタム信号

八、マルチappアプリケーション(青写真で実現)

要求コンテキストソース分析、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  

プログラム実行、2つのLocalStack()オブジェクト、1つはrequestとsession、もう1つはgとcurrent_app二、gオブジェクト
1、gオブジェクトとは
ユーザ情報を格納するために特化したgオブジェクト,gの全称global
gオブジェクトは,一度のリクエストにおけるすべてのコードの場所で使用可能である.
2、gオブジェクトとセッションの違い

session      request ，  session    ，   request          session
  g    ，g          ，     g        ，         

三、flask-session
1、作用
デフォルトで保存されている署名クッキーの値をredis/memcached/file/Mongodb/SQLAlchemyに保存
2、取り付け

pip3 install flask-session

3、使用
(1)redisでセッションを保存する

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)構成変更による挿抜可能保存セッションの実現

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)

質問:クッキーを設定する場合、ブラウザを閉じるとクッキーが無効になるように設定します.

response.set_cookie('k','v',exipre=None)    #       
#  session   
app.session_interface=RedisSessionInterface(conn,key_prefix='lqz',permanent=False)
#     ，           ，       

質問:クッキーのデフォルトタイムアウト時間はいくらですか?タイムアウト時間の設定方法

#   expires = self.get_expiration_time(app, session)
'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),  #         

 
四、flask-script
1、flaskとは何かscript
djangoのpython 3 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-z\d$@$!%*?&]{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

<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Titletitle>
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()

<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Titletitle>
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)

 
ブログの内容はただ参考にして、一部は他人の優秀な博文を参考にして、ただ学習のために使用します