目錄

• 請求上下文源碼分析、g對象、第三方插件（flask_session、flask_script、wtforms）、信號
  • 一、請求上下文源碼分析
  • 二、g對象
    • 1、什麽是g對象
    • 2、g對象和session的區別
  • 三、flask-session
    • 1、作用
    • 2、安裝
    • 3、使用
  • 四、flask-script
    • 1、什麽是flask_script
    • 2、安裝
    • 3、使用
    • 4、自定義命令
  • 五、數據庫連接池
    • 1、pymsql鏈接數據庫
    • 2、數據庫連接池版
  • 六、wtforms
    • 1、安裝
    • 2、使用
  • 七、信號
    • 1、什麽是信號
    • 2、安裝
    • 3、內置信號
    • 4、使用信號
    • 5、一個流程中的信號觸發點
    • 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刪除
 

程序運行，兩個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-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

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

Flask框架 （四）—— 請求上下文源碼分析、g對象、第三方插件（flask_session、flask_script、wtforms）、信號