Flask簡介
Flask是主流PythonWeb三大框架之一，其特點是短小精悍以及功能強大從而獲得眾多Pythoner的追捧，相比於Django它更加簡單更易上手，Flask擁有非常強大的三方庫，提供各式各樣的模塊對其本身進行擴充：
下面是Flask與Django本身的一些區別：
| Flask | Django | |
|---|---|---|
| 網關接口(WSGI) | werkzeug | wsgiref |
| 模板語言(Template) | Jinja2 | DjangoTemplate |
| ORM | SQLAlchemy | DjangoORM |
下載Flask：
pip install flask
werkzeug模塊
Flask本質就是對werkzeug模塊進行一些更高層次的封裝，就如同Django是對wsgiref模塊做了一些更高層次封裝一樣。所以先來看一下werkzeug模塊如何使用：
from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple
@Request.application
def index(request):
return Response("Hello werkzeug")
if __name__ == '__main__':
run_simple("localhost", 5000, index)
簡單入門
基本使用
使用Flask的小案例：
from flask import Flask
# 1.創建Flask對象實例，填入構造參數
app = Flask(__name__)
# 2.視圖函數中書寫route以及view
@app.route("/")
def index():
return "Hello Flask!"
# 3.啟動監聽，等待鏈接請求 默認端口號：5000，可在run()時添加形參
if __name__ == '__main__':
app.run()
# app.run(Thread=True) # 開啟多線程
構造參數
對於創建Flask實例對象，傳入的構造參數有以下選項：
| 形參 | 描述 | 默認值 |
|---|---|---|
| import_name | 為Flask對象取名，一般為__name__即可 | 無 |
| static_url_path | 模板中訪問的靜態文件存放目錄，默認情況下與static_folder同名 | None |
| static_folder | 靜態文件存放的目錄名稱，默認當前項目中的static目錄 | static |
| static_host | 遠程靜態文件所用的Host地址 | None |
| host_matching | 如果不是特別需要的話,慎用,否則所有的route都需要host=""的參數 | False |
| subdomain_matching | SERVER_NAME子域名，暫時未GET到其作用 | False |
| template_folder | template模板目錄, 默認當前項目中的templates目錄 | templates |
| instance_path | 指向另一個Flask實例的路徑 | None |
| instance_relative_config | 是否加載另一個實例的配置 | False |
| root_path | 主模塊所在的目錄的絕對路徑，默認項目目錄 | None |
Flask配置項
如同Django中的settings.py一樣，在Flask中也擁有它自己的一些配置項。通過以下方式可對配置項進行修改。
debug模式
一般來說對於Flask的開發模式都是用app.debug=True來完成的：
app = Flask(__name__)
app.debug = True
當然你也可以依照下面的方式進行修改。
config修改
對Flask實例直接進行config的字典操作修改配置項：
app = Flask(__name__)
app.config["DEBUG"] = True
from_pyfile
以py文件形式進行配置：
app = Flask(__name__)
app.config.from_pyfile("flask_settings.py")
# flask_settings.py
DEBUG = True
from_object
以class與類屬性的方式書寫配置項：
app = Flask(__name__)
app.config.from_object("flask_settings.DevelopmentConfig")
# flask_settings.py
class BaseConfig(object):
"""
抽象類，只用於繼承
"""
DEBUG = False
TESTING = False
# 其他配置項
class ProductionConfig(BaseConfig):
"""
上線時的配置項
"""
DATABASE_URI = 'mysql://user@localhost/foo'
class DevelopmentConfig(BaseConfig):
"""
開發時的配置項
"""
DEBUG = True
其他配置
通過環境變量配置：
app.config.from_envvar("環境變量名稱")
# 環境變量的值為python文件名稱名稱，內部調用from_pyfile方法
通過JSON格式文件配置：
app.config.from_json("json文件名稱")
# JSON文件名稱，必須是json格式，因為內部會執行json.loads
通過字典格式配置：
app.config.from_mapping({'DEBUG':True})
配置項大全
以下是Flask的配置項大全：
'DEBUG': False, # 是否開啟Debug模式
'TESTING': False, # 是否開啟測試模式
'PROPAGATE_EXCEPTIONS': None, # 異常傳播(是否在控制台打印LOG) 當Debug或者testing開啟后,自動為True
'PRESERVE_CONTEXT_ON_EXCEPTION': None, # 一兩句話說不清楚,一般不用它
'SECRET_KEY': None, # 之前遇到過,在啟用Session的時候,一定要有它
'PERMANENT_SESSION_LIFETIME': 31, # days , Session的生命周期(天)默認31天
'USE_X_SENDFILE': False, # 是否棄用 x_sendfile
'LOGGER_NAME': None, # 日志記錄器的名稱
'LOGGER_HANDLER_POLICY': 'always',
'SERVER_NAME': None, # 服務訪問域名
'APPLICATION_ROOT': None, # 項目的完整路徑
'SESSION_COOKIE_NAME': 'session', # 在cookies中存放session加密字符串的名字
'SESSION_COOKIE_DOMAIN': None, # 在哪個域名下會產生session記錄在cookies中
'SESSION_COOKIE_PATH': None, # cookies的路徑
'SESSION_COOKIE_HTTPONLY': True, # 控制 cookie 是否應被設置 httponly 的標志，
'SESSION_COOKIE_SECURE': False, # 控制 cookie 是否應被設置安全標志
'SESSION_REFRESH_EACH_REQUEST': True, # 這個標志控制永久會話如何刷新
'MAX_CONTENT_LENGTH': None, # 如果設置為字節數， Flask 會拒絕內容長度大於此值的請求進入，並返回一個 413 狀態碼
'SEND_FILE_MAX_AGE_DEFAULT': 12, # hours 默認緩存控制的最大期限
'TRAP_BAD_REQUEST_ERRORS': False,
# 如果這個值被設置為 True ，Flask不會執行 HTTP 異常的錯誤處理，而是像對待其它異常一樣，
# 通過異常棧讓它冒泡地拋出。這對於需要找出 HTTP 異常源頭的可怕調試情形是有用的。
'TRAP_HTTP_EXCEPTIONS': False,
# Werkzeug 處理請求中的特定數據的內部數據結構會拋出同樣也是“錯誤的請求”異常的特殊的 key errors 。
# 同樣地，為了保持一致，許多操作可以顯式地拋出 BadRequest 異常。
# 因為在調試中，你希望准確地找出異常的原因，這個設置用於在這些情形下調試。
# 如果這個值被設置為 True ，你只會得到常規的回溯。
'EXPLAIN_TEMPLATE_LOADING': False,
'PREFERRED_URL_SCHEME': 'http', # 生成URL的時候如果沒有可用的 URL 模式話將使用這個值
'JSON_AS_ASCII': True,
# 默認情況下 Flask 使用 ascii 編碼來序列化對象。如果這個值被設置為 False ，
# Flask不會將其編碼為 ASCII，並且按原樣輸出，返回它的 unicode 字符串。
# 比如 jsonfiy 會自動地采用 utf-8 來編碼它然后才進行傳輸。
'JSON_SORT_KEYS': True,
#默認情況下 Flask 按照 JSON 對象的鍵的順序來序來序列化它。
# 這樣做是為了確保鍵的順序不會受到字典的哈希種子的影響，從而返回的值每次都是一致的，不會造成無用的額外 HTTP 緩存。
# 你可以通過修改這個配置的值來覆蓋默認的操作。但這是不被推薦的做法因為這個默認的行為可能會給你在性能的代價上帶來改善。
'JSONIFY_PRETTYPRINT_REGULAR': True,
'JSONIFY_MIMETYPE': 'application/json',
'TEMPLATES_AUTO_RELOAD': None,
路由
路由參數
所有路由中的參數如下：
@app.route("/index", methods=["POST", "GET"], endpoint="別名", defaults={"默認參數": 1}, strict_slashes=True,
redirect_to="/", subdomain=None)
詳細描述：
| 參數 | 描述 |
|---|---|
| methods | 訪問方式，默認只支持GET |
| endpoint | 別名、默認為函數名，不可重復。默認為函數名 |
| defaults | 當視圖函數擁有一個形參時，可將它作為默認參數傳遞進去 |
| strict_slashes | 是否嚴格要求路徑訪問，如定義的時候是/index，訪問是/index/，默認是嚴格訪問 |
| redirect_to | 301永久重定向，如函數help的redirect_to是/doc，則訪問help將跳轉到doc函數 |
| subdomain | 通過指定的域名進行訪問，在瀏覽器中輸入域名即可，本地需配置hosts文件 |
轉換器
Flask中擁有Django3中的轉換器來捕捉用戶請求的地址欄參數：
| 轉換器 | 含義 |
|---|---|
| default | 接收字符串，默認的轉換器 |
| string | 接收字符串，和默認的一樣 |
| any | 可以指定多個路徑 |
| int | 接收整數 |
| float | 接收浮點數和整數 |
| uuid | 唯一標識碼 |
| path | 和字符串一樣,但是它可以配置/，字符串不可以 |
如下所示：
http://localhost:5000/article/2020-01-29
@app.route("/article/<int:year>-<int:month>-<int:day>", methods=["POST", "GET"])
def article(year, month, day):
# 相當於有命分組，必須使用同樣的變量名接收
# 並且還會自動轉換類型，int捕捉到的就都是int類型
return f"{year}-{month}-{day}"
正則匹配
由於參數捕捉只支持轉換器，所以我們可以自定義一個轉換器讓其能夠支持正則匹配：
from flask import Flask, url_for
from werkzeug.routing import BaseConverter
app = Flask(import_name=__name__)
class RegexConverter(BaseConverter):
"""
自定義URL匹配正則表達式
"""
def __init__(self, map, regex):
super(RegexConverter, self).__init__(map)
self.regex = regex
def to_python(self, value):
"""
路由匹配時，匹配成功后傳遞給視圖函數中參數的值
:param value:
:return:
"""
return int(value)
def to_url(self, value):
"""
使用url_for反向生成URL時，傳遞的參數經過該方法處理，返回的值用於生成URL中的參數
:param value:
:return:
"""
val = super(RegexConverter, self).to_url(value)
return val
# 添加到flask中
app.url_map.converters['regex'] = RegexConverter
@app.route('/index/<regex("\d+"):nid>')
def index(nid):
print(url_for('index', nid='888'))
return 'Index'
if __name__ == '__main__':
app.run()
反向解析
使用url_for()可在視圖中反向解析出url：
# url_for(endpoint, **values)
print(url_for("article", **{"year": 2010, "month": 11, "day": 11}))
print(url_for("article", year=2010, month=11, day=11))
如果在模板中，也可以使用url_for()進行反向解析：
<a href='{{ url_for("article", year=2010, month=11, day=11)) }}'>點我</a>
app.add_url_rule
可以發現，Flask的路由與Django的有非常大的區別，但是通過app.add_url_rule也可以做到和Django相似。
但是這樣的做法很少，函數簽名如下：
def add_url_rule(
self,
rule, # 規則
endpoint=None, # 別名
view_func=None, # 視圖函數
provide_automatic_options=None, # 控制是否自動添加options
**options
):
實際應用如下：
from flask import Flask
app = Flask(__name__)
def index():
return "index"
def home(name):
return "Welcome Home, %s" % name
routers = [
("/index", None, index),
("/home/<string:name>", None, home),
]
for rule in routers:
app.add_url_rule(*rule)
if __name__ == '__main__':
app.run()
視圖
請求相關
Flask的request對象不是通過參數傳遞，而是通過導入：
from flask import request
下面是一些常用的屬性與方法：
| 屬性/方法 | 描述 |
|---|---|
| request.headers | 查看所有的請求頭 |
| request.method | 存放請求方式 |
| request.form | 存放form表單中的序列化數據，一般來說就是POST請求的數據 |
| request.args | 存放url里面的序列化數據，一般來說就是GET請求的數據 |
| request.data | 查看傳過來所有解析不了的內容 |
| request.json | 查看前端傳過來的json格式數據，內部會自動反序列化 |
| request.values.to_dict() | 存放url和from中的所有數據 |
| request.cookies | 前端傳過來的cookies |
| request.path | 路由地址，如：/index |
| request.full_path | 帶參數的請求路由地址，如：/index?name=yunya |
| request.url | 全部地址，如：http://127.0.0.1:5000/index?name=yunya |
| request.host | 主機位，如：127.0.0.1:5000 |
| request.host_url | 將主機位轉換成url，如：http://127.0.0.1:5000/ |
| request.url_root | 域名 |
| file = request.files | 前端傳過來的文件 |
| file.filename | 返回文件名稱 |
| file.save() | 保存文件 |
操作演示：
from flask import Flask
from flask import request
app = Flask(__name__)
@app.route('/index',methods=["POST","GET"])
def index():
print(request.method)
if request.method == "GET":
return "GET"
elif request.method == "POST":
return "POST"
else:
return "ERROR"
if __name__ == '__main__':
app.run()
返回響應
返回響應一般有五種：
| 返回響應 | 描述 |
|---|---|
| return 'string' | 返回字符串 |
| return render_template() | 返回模板文件 |
| return redirect() | 302，重定向，可填入別名或者路由匹配地址 |
| return jsonify() | 返回Json格式數據 |
| return Response對象 | 直接返回一個對象，常用於取消XSS攻擊預防、設置返回頭等 |
注意，在Flask中，都會返回csrftoken，它存放在瀏覽器的cookie中。當Flask模板渲染的頁面發送請求時會自動攜帶csrftoken，這與Django並不相同
此外，如果返回的對象不是字符串、不是元組也不是Response對象，它會將值傳遞給Flask.force_type類方法，將它轉換成為一個響應對象
如下所示：
from flask import Flask
app = Flask(__name__)
@app.route('/templateTest')
def templateTest():
# 返回模板
from flask import render_template
return render_template("result.html")
@app.route('/redirectTest')
def redirectTest():
# 302重定向
from flask import redirect
return redirect("templateTest")
@app.route('/jsonTest')
def jsonTest():
# 返回json數據
from flask import jsonify
message = {"book": "flask", "price": 199, "publish": "BeiJing"}
return jsonify(message)
@app.route('/makeResponseTest')
def makeResponseTest():
# 返回Response對象
from flask import make_response
# 取消XSS攻擊預防
from flask import Markup
element = Markup("<a href='https://www.google.com'>點我一下</a>")
response = make_response(element)
# 操作cookie
response.set_cookie("key", "oldValue")
response.delete_cookie("key")
response.set_cookie("key", "newValue")
# 操作返回頭
response.headers["jwt"] = "ajfkdasi#@#kjdfsas9f(**jfd"
return response
if __name__ == '__main__':
app.run()
session
在Flask中，session也是通過導入來操縱的，而不是通過request對象。
需要注意的是在Flask中`session的保存時長為31天，並且默認是保存在內存中，並未做任何持久化處理。
如果想做持久化處理，則可以通過其他的一些第三方模塊。
| 操作 | 描述 |
|---|---|
| session.get("key",None) | 獲取session |
| session["key"]=value | 設置session |
| session.pop("key",None) | 刪除session |
如下案例所示：
from flask import Flask
from flask import request
from flask import session
from flask import Markup
from flask import render_template
from flask import redirect
# 第一步，導入session
app = Flask(__name__)
# 第二步，加鹽，也可以在配置文件中加鹽
app.secret_key = "salt"
@app.route('/home')
def home():
username = session.get("username")
print(username)
if username:
return "歡迎回家%s"%username
return redirect("login")
@app.route('/login',methods=["GET","POST"])
def login():
if request.method == "GET":
return render_template("login.html")
if request.method == "POST":
username = request.form.get("username")
if username:
session["username"] = username
return "您已登錄" + Markup("<a href='/home'>返回home</a>")
return redirect("login")
if __name__ == '__main__':
app.run()
<form method="POST">
<p><input type="text" name="username" placeholder="username"></p>
<p><input type="text" name="password" placeholder="password"></p>
<button type="submit">登錄</button>
</form>
flash
消息閃現flash是基於session來做的，它只會允許值被取出一次，內部通過pop()實現。
使用方式如下：
flash("data", category="sort")
# 存入數據以及分類
get_flashed_messages(with_categories=False, category_filter=())
# 取出flash中的數據
# with_categories為True時返回一個tuple
# category_filter指定數據類別，如不指定則代表取出所有
如下所示：
from flask import Flask
from flask import flash
from flask import get_flashed_messages
app = Flask(__name__)
app.secret_key = "salt"
@app.route('/set_flash')
def set_flash():
flash(message="dataA",category="sortA")
flash(message="dataB",category="sortB")
return "OK!!"
@app.route('/get_flash/<string:choice>')
def get_flash(choice):
if choice == "all":
all_data = get_flashed_messages() # 取所有閃現消息
return str(all_data) # ['dataA', 'dataB']
elif choice == "sortA":
sortA_data = get_flashed_messages(category_filter=("sortA",)) # 取類別A的所有閃現消息
return str(sortA_data) # ['dataA']
elif choice == "sortB":
sortB_data = get_flashed_messages(category_filter=("sortB",)) # 取類別B的所有閃現消息
return str(sortB_data) # ['dataB']
else:
return "ERROR"
if __name__ == '__main__':
app.run()
FBV
如果不是做前后端分離，那么Flask應用最多的還是FBV：
from flask import Flask
app = Flask(__name__)
@app.route('/index')
def index():
return "index"
if __name__ == '__main__':
app.run()
CBV
使用CBV必須導入views.MethodView且繼承它，初此之外必須使用app.add_url_rule添加路由與視圖的關系映射：
from flask import Flask
from flask.views import MethodView
app = Flask(__name__)
class Home(MethodView):
methods = ["GET", "POST"] # 該類中允許的請求方式
decorators = [] # 裝飾器添加在這里
def dispatch_request(self, *args, **kwargs):
print("首先執行該方法")
return super(Home, self).dispatch_request(*args, **kwargs)
def get(self):
return "Home,Get"
def post(self):
return "Home,Post"
app.add_url_rule("/home",view_func=Home.as_view(name="home"))
if __name__ == '__main__':
app.run()
RESTAPI
如果項目是前后端分離的，則需要借助第三方模塊flask-restful，詳情查閱官網：
文件上傳案例
保存上傳文件的案例：
from flask import Flask, request
app = Flask(__name__)
'''因為是文件，所以只能是POST方式'''
@app.route("/upload", methods=["POST"])
def upload():
"""接受前端傳送來的文件"""
file_obj = request.files.get("pic")
if file_obj is None:
# 表示沒有發送文件
return "未上傳文件"
'''
將文件保存到本地（即當前目錄）
直接使用上傳的文件對象保存
'''
file_obj.save('pic.jpg') # 和前端上傳的文件類型要相同
return "上傳成功"
# 將文件保存到本地（即當前目錄） 普通的保存方法
# with open("./pic.jpg",'wb') as f:
# data = file_obj.read()
# f.write(data)
# return "上傳成功"
if __name__ == '__main__':
app.run(debug=True)
其他的一些補充知識：
file_obj.stream # 文件流，即文件的二進制對象
from werkzeug.datastructures import FileStorage # 查看詳情，文件對象的具體方法
模板
jinja2簡介
jinja2是Flask中默認的模板語言，相比於DjangoTemplate它更加的符合Python語法。
如在模板傳參中，如果視圖中傳入是一個dict，那么在DTL中只能通過.的方式進行深度獲取，而在jinja2中則可以通過[]的方式進行獲取。
此外，在DTL中如果視圖傳入一個function則會自動加括號進行調用，而在jinja2中就不會進行自動調用而是要自己手動加括號進行調用。
總而言之，jinja2相比於DTL來說更加的人性化。
模板傳參
模板傳參可以通過k=v的方式傳遞，也可以通過**dict的方式進行解包傳遞：
@app.route('/index')
def index():
context = {
"name": "雲崖",
"age": 18,
"hobby": ["籃球", "足球"]
}
return render_template("index.html", **context)
# return render_template("index.html", name="雲崖", age=18)
渲染，通過{{}}進行：
<body>
<p>{{name}}</p>
<p>{{age}}</p>
<p>{{hobby.0}}-{{hobby[1]}}</p>
</body>
內置過濾器
常用的內置過濾器如下：
| 過濾器 | 描述 |
|---|---|
| escape | 轉義字符 |
| safe | 關閉XSS預防，關閉轉義 |
| striptags | 刪除字符串中所有的html標簽，如果有多個空格連續，將替換為一個空格 |
| first | 返回容器中的第一個元素 |
| last | 返回容器中的最后一個元素 |
| length | 返回容器總長度 |
| abs | 絕對值 |
| int | 轉換為int類型 |
| float | 轉換為float類型 |
| join | 字符串拼接 |
| lower | 轉換為小寫 |
| upper | 轉換為大寫 |
| capitialize | 把值的首字母轉換成大寫，其他子母轉換為小寫 |
| title | 把值中每個單詞的首字母都轉換成大寫 |
| trim | 把值的首尾空格去掉 |
| round | 默認對數字進行四舍五入，也可以用參數進行控制 |
| replace | 替換 |
| format | 格式化字符串 |
| truncate | 截取length長度的字符串 |
| default | 相當於or，如果渲染變量沒有值就用default中的值 |
使用內置過濾器：
<p>{{gender | default("性別不詳")}}</p>
分支循環
if和for都用{% %}進行包裹，與DTL中使用相似。
在for中擁有以下變量，用來獲取當前的遍歷狀態：
| for循環的遍歷 | 描述 |
|---|---|
| loop.index | 當前遍歷次數，從1開始計算 |
| loop.index0 | 當前遍歷次數，從0開始計算 |
| loop.first | 第一次遍歷 |
| loop.last | 最后一次遍歷 |
| loop.length | 遍歷對象的長度 |
| loop.revindex | 到循環結束的次數，從1開始 |
| loop.revindex0 | 到循環結束的次數，從0開始 |
下面是一則示例：
<body>
{% for item in range(10) %}
{% if loop.first %}
<p>第一次遍歷開始--->{{loop.index}}</p>
{% elif loop.last %}
<p>最后一次遍歷開始-->{{loop.index}}</p>
<p>遍歷了一共{{loop.length}}次</p>
{% else %}
<p>{{loop.index}}</p>
{% endif %}
{% endfor %}
</body>
結果如下：
第一次遍歷開始--->1
2
3
4
5
6
7
8
9
最后一次遍歷開始-->10
遍歷了一共10次
宏的使用
在模板中的宏類似於Python中的函數，可對其進行傳值：
<body>
<!--定義宏，后面是默認的參數-->
{% macro input(name, value="", type="text") %}
<input name="{{ name }}" value="{{ value }}" type="{{ type }}">
{% endmacro %}
<!--使用宏-->
<form action="">
<p>username:{{ input("username") }}</p>
<p>password:{{ input("pwd", type="password")}}</p>
<p>{{ input(value="login", type="submit") }}</p>
</form>
</body>
可以在一個模板中專門定義宏，其他模板中再進行導入：
# 導入方式一
# with context可以把后端傳到當前模板的變量傳到定義的宏里面
{% import "macros.html" as macro with context %}
<form>
<p>用戶名：{{ macro.input('username') }}</p>
<p>密碼：{{ macro.input('password',type="password" )}}</p>
<p> {{ macro.input(value="提交",type="submit" )}}</p>
</form>
# 導入方式二
{% from "macros.html" import input as input_field %}
<form>
<p>用戶名：{{ input_field('username') }}</p>
<p>密碼：{{ input_field('password',type="password" )}}</p>
<p> {{ input_field(value="提交",type="submit" )}}</p>
</form>
定義變量
在模板中可通過{% set %}和{% with %}定義變量。
{% set %}是全局變量，可在當前模板任意位置使用
{% with %}是局部變量，只能在{% with %}語句塊中使用
<body>
{% set name="雲崖" %}
<p>名字是:{{name}}</p>
{% with age=18 %}
<p>年齡是:{{age}}</p>
{% endwith %}
</body>
模板繼承
使用{% extends %}引入一個定義號的模板。
使用{% blocak %}和{% endblock %}定義塊
使用{{ super() }}引入原本的模板塊內容
定義模板如下：
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport"
content="width=device-width, user-scalable=no, initial-scale=1.0, maximum-scale=1.0, minimum-scale=1.0">
<meta http-equiv="X-UA-Compatible" content="ie=edge">
<title>jinja2學習</title>
{% block style %}
{% endblock %}
</head>
<body>
<header>
{% block header %}
<p>頭部信息</p>
{% endblock %}
</header>
<main>
{% block main %}
<p>主體信息</p>
{% endblock %}
</main>
<footer>
<p>頁面尾部</p>
</footer>
{% block script %}
{% endblock %}
</body>
</html>
導入模板並使用：
{% extends "base.html" %}
{% block style %}
<style>
h1{
color:red;
}
</style>
{% endblock %}
{% block header %}
<!--調用父模板內容-->
{{ super() }}
{% endblock %}
{% block main %}
<h1>HELLO,歡迎來到Jinja2學習</h1>
{% endblock %}
{% block script %}
<script>
"use strict;"
console.log("HELLO,WORLD")
</script>
{% endblock %}
中間件
在Flask中的中間件使用非常少。由於Flask是基於werkzeug模塊來完成的，所以按理說我們只需要在werkzeug的啟動流程中添加代碼即可。
下面是中間件的使用方式，如果想了解它的原理在后面的源碼分析中會有涉及。
在Flask請求來臨時會執行wsgi_app這樣的一個方法，所以就在這個方法上入手：
from flask import Flask
from flask import render_template
app = Flask(__name__)
# 中間件
class Middleware(object):
def __init__(self, old_wsgi_app):
self.old_wsgi_app = old_wsgi_app # 原本要執行的wsgi_app方法
def __call__(self, environ, start_response):
print("書寫代碼...中間件。請求來時")
result = self.old_wsgi_app(environ, start_response)
print("書寫代碼...中間件。請求走時")
return result
@app.route('/index')
def index():
return "Hello,world"
if __name__ == '__main__':
app.wsgi_app = Middleware(app.wsgi_app) # 傳入要原本執行的wsgi_app
app.run()
裝飾器
如何添加裝飾器
由於Flask的每個視圖函數頭頂上都有一個裝飾器，且具有endpoint不可重復的限制。
所以我們為單獨的某一個視圖函數添加裝飾器時一定要將其添加在下方（執行順序自下而上），此外還要使用functools.wraps修改裝飾器inner()讓每個裝飾器的inner.__name__都不相同，來突破endpoint不可重復的限制。
如下所示，為單獨的某一個接口書寫頻率限制的裝飾器：
from flask import Flask
from functools import wraps
app = Flask(__name__)
def flow(func):
@wraps(func) # 如果不加這個，路由的別名一致都是inner就會拋出異常。func.__name__
def inner(*args,**kwargs):
# 書寫邏輯，用random代替。如果是0就代表不讓通過
import random
access = random.randint(0,1)
if access:
result = func(*args,**kwargs)
return result
else:
return "頻率太快了"
return inner
@app.route('/backend')
@flow
def backend():
return "backend"
@app.route('/index')
@flow
def index():
return "index"
if __name__ == '__main__':
app.run()
befor_request
每次請求來的時候都會走它，由於Flask的中間件比較弱雞，所以這種方式更常用。
類似於Django中間件中的process_request，如果有多個順序是從上往下，可以用它做session認證。
如果返回的不是None，就攔截請求
@app.before_request
def before(*args,**kwargs):
if request.path=='/login':
return None
else:
name=session.get('user')
if not name:
return redirect('/login')
else:
return None
after_request
請求走了就會觸發,類似於Django的process_response，如果有多個，順序是從下往上執行：
必須傳入一個參數，就是視圖的return值
@app.after_request
def after(response):
print('我走了')
return response
before_first_request
目啟動起來第一次會走，以后都不會走了，也可以配多個（項目啟動初始化的一些操作）
如果返回的不是None，就攔截請求
@app.before_first_request
def first():
print('我的第一次')
teardown_request
每次視圖函數執行完了都會走它。
可以用來記錄出錯日志：
@app.teardown_request
def ter(e):
print(e)
print('我是teardown_request ')
errorhandler
綁定錯誤的狀態碼，只要碼匹配就走它。
常用於重寫404頁面等：
@app.errorhandler(404)
def error_404(arg):
return render_template('error.html',message='404錯誤')
template_global
定義全局的標簽，如下所示：
@app.template_global()
def add(a1, a2):
return a1 + a2
# 在模板中：{{ add(3,4) }}
template_filter
定義全局過濾器，如下所示：
@app.template_filter()
def db(a1, a2, a3): # 第一個值永遠都是|左邊的值
return a1 + a2 + a3
# 在模板中{{ 1|db(2,3)}}
多request順序
如果存在多個berfor_request與多個after_request那么執行順序是怎樣的？
from flask import Flask
from functools import wraps
app = Flask(__name__)
@app.before_request
def before_fist():
print("第一個before_request")
@app.before_request
def before_last():
print("第二個before_request")
@app.after_request
def before_fist(response):
print("第一個after_request")
return response
@app.after_request
def before_last(response):
print("第二個after_request")
return response
@app.route('/index')
def index():
return "index"
if __name__ == '__main__':
app.run()
"""
第一個before_request
第二個before_request
第二個after_request
第一個after_request
"""
如果第一個before_request就返回了非None進行攔截，執行順序則和Django的不一樣，Django會返回同級的process_response，而Flask還必須要走所有的after_request的：
@app.before_request
def before_fist():
print("第一個before_request")
return "攔截了"
第一個before_request
第二個after_request
第一個after_request
藍圖
藍圖Blueprint的作用就是為了將功能和主服務分開。
說的直白點就是構建項目目錄，划分內置的裝飾器作用域等，類似於Django中app的概念。
小型項目
下面有一個基本的項目目錄，如下所示：
- mysite
- mysite # 包，項目根目錄
- views # 文件夾，視圖相關
- index.py
- backend.py
- templates
- index # 文件夾，index相關的模板
- backend # 文件夾，backend相關的資源
- static
- index
- backend
- __init__.py
- manage.py # 啟動文件
- settings.py # 配置文件
這樣的目錄結構看起來就比較清晰，那么如何對它進行管理呢？就可以使用藍圖：
# backend.py
from flask import Blueprint
from flask import render_template
bck = Blueprint("bck", __name__) # 創建藍圖對象 bck
@bck.route("/login") # 路由使用藍圖對象bck為前綴，而不是app
def login():
return render_template("backend/backend_login.html")
# index.py
from flask import Blueprint
from flask import render_template
idx = Blueprint("idx", __name__)
@idx.route("/login")
def login():
return render_template("index/index_login.html")
# __init__.py
from flask import Flask
from .views.backend import bck
from .views.index import idx
def create_app():
app = Flask(import_name=__name__)
app.register_blueprint(bck) # 注冊藍圖對象 bck
app.register_blueprint(idx) # 注冊藍圖對象 idx
return app
# manage.py
from mysite import create_app
if __name__ == '__main__':
app = create_app()
app.run()
url前綴
啟動服務后發現兩個功能區的login都是相同的url，導致后注冊的藍圖對象永遠無法訪問登錄頁面。
在__init__.py中注冊藍圖對象的代碼中添加前綴：
from flask import Flask
from .views.backend import bck
from .views.index import idx
def create_app():
app = Flask(import_name=__name__)
app.register_blueprint(bck, url_prefix="/backend/")
app.register_blueprint(idx, url_prefix="/index/")
return app
訪問時：
http://127.0.0.1:5000/backend/login
http://127.0.0.1:5000/index/login
藍圖資源
每個藍圖應用的資源都不相同，如下：
templates/index # 這是index訪問的模板路徑
templates/backend # 這是backend訪問的模板路徑
static/index
static/backend
如何指定他們的資源呢？其實在創建藍圖對象的時候就可以指定：
from flask import Blueprint
from flask import render_template
# 使用相對路徑
bck = Blueprint("bck",__name__, template_folder="../templates/backend", static_folder="../static/backend",)
@bck.route("/login")
def login():
return render_template("backend_login.html") # 注意不同藍圖對象之間的模板應該盡量不重名，重名可能導致一些錯誤
如果是靜態資源的訪問，並不會加上前綴/backend：
<img src="/static/backend/logo@2x.png" alt="">
藍圖裝飾器
藍圖裝飾器分為全局裝飾器和局部裝飾器兩種：
全局裝飾器全局有效：
def create_app():
app = Flask(import_name=__name__)
app.register_blueprint(bck, url_prefix="/backend/",)
app.register_blueprint(idx, url_prefix="/index/")
@app.before_request
def func():
print("全局有效")
return app
局部裝飾器只在當前藍圖對象bck有效：
bck = Blueprint("bck",__name__, template_folder="../templates/backend",static_folder="../static/backend",)
@bck.before_request
def func():
print("局部有效")
大型項目
構建大型項目，就完全可以將它做的和Django相似，讓每個藍圖對象都擁有自己的templates與static。
- mysite
- mysite
- index # 包，單獨的一個藍圖對象
- static # 文件夾
- templates # 文件夾
- views.py
- __init__.py # 創建藍圖對象，指定template與static
- backend
- static
- templates
- views.py
- __init__.py
- manage.py # 啟動文件
- settings.py # 配置文件
多app應用
一個Flask程序允許多個實例，如下所示：
from werkzeug.wsgi import DispatcherMiddleware
from werkzeug.serving import run_simple
from flask import Flask
app01 = Flask('app01')
app02 = Flask('app02')
@app01.route('/index')
def index():
return "app01"
@app02.route('/index')
def index2():
return "app02"
app = DispatcherMiddleware(app01, {
'/app01': app01,
'/app02': app02,
})
#默認使用app01的路由，也就是訪問 http://127.0.0.1:5000/index 返回app01
#當以app01開頭時候使用app01的路由，也就是http://127.0.0.1:5000/app01/index 返回app01
#當以app02開頭時候使用app02的路由，也就是http://127.0.0.1:5000/app02/index 返回app02
if __name__ == "__main__":
run_simple('127.0.0.1', 5000, app)
解決跨域
解決跨域請求，可以用第三方插件，也可以自定義響應頭：
@app.after_request # 解決CORS跨域請求
def cors(response):
response.headers['Access-Control-Allow-Origin'] = "*"
if request.method == "OPTIONS":
response.headers["Access-Control-Allow-Headers"] = "Origin,Content-Type,Cookie,Accept,Token,authorization"
return response
上下文機制
全局變量
在Flask項目啟動時，會自動初始化一些全局變量。其中有幾個變量尤為重要，可通過以下命令查看：
from flask import globals
就是下面的6個變量，將貫穿整個HTTP請求流程。
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, "request")) # from flask import request 拿的就是它
session = LocalProxy(partial(_lookup_req_object, "session")) # from flask import session 拿的就是它
g = LocalProxy(partial(_lookup_app_object, "g"))
偏函數
上面的6個變量中有兩個變量執行了類的實例化，並且有傳入了一個偏函數。
偏函數的作用在於不用傳遞一個參數，設置好后自動傳遞：
from functools import partial
def add(x, y):
return x + y
add = partial(add,1) # 自動傳遞第一個參數為1，返回一個新的函數
result = add(2)
print(result) # 3
列表實現棧
棧是一種后進先出的數據結構，使用列表可以實現一個棧：
class Stack(object):
def __init__(self):
self.__stack = []
def push(self, value):
self.__stack.append(value)
@property
def top(self):
try:
return self.__stack[-1]
except IndexError as e:
return None
stack = Stack()
stack.push(1)
print(stack.top)
在Flask源碼中多次有構建這個棧的地方（目前來看至少兩處）。
Local
Local對象在全局變量中會實例化兩次，作用是實例化出一個字典，用於存放線程中的東西：
_request_ctx_stack = LocalStack()
_app_ctx_stack = LocalStack()
我們來看看它的源碼：
try: # 導入協程獲取pid，或者是線程模塊獲取pid的函數
from greenlet import getcurrent as get_ident
except ImportError:
try:
from thread import get_ident
except ImportError:
from _thread import get_ident
class Local(object):
# 只能 . 這里面的
__slots__ = ("__storage__", "__ident_func__")
def __init__(self):
object.__setattr__(self, "__storage__", {})
object.__setattr__(self, "__ident_func__", get_ident)
# 返回可迭代對象，這里可以看出__storage__是一個字典
def __iter__(self):
return iter(self.__storage__.items())
def __call__(self, proxy):
return LocalProxy(self, proxy)
def __release_local__(self):
self.__storage__.pop(self.__ident_func__(), None)
# 通過pid返回字典中的一個name對應的value
def __getattr__(self, name):
try:
return self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
# 構建出一個字典，{pid:{name:value}}
def __setattr__(self, name, value):
ident = self.__ident_func__()
storage = self.__storage__
try:
storage[ident][name] = value
except KeyError:
storage[ident] = {name: value}
def __delattr__(self, name):
try:
del self.__storage__[self.__ident_func__()][name]
except KeyError:
raise AttributeError(name)
LocalStack
用於操縱Local中構建的字典：
class LocalStack(object):
# 實例化
def __init__(self):
self._local = Local()
def __release_local__(self):
self._local.__release_local__()
@property
def __ident_func__(self):
return self._local.__ident_func__
@__ident_func__.setter
def __ident_func__(self, value):
object.__setattr__(self._local, "__ident_func__", value)
def __call__(self):
def _lookup():
rv = self.top
if rv is None:
raise RuntimeError("object unbound")
return rv
return LocalProxy(_lookup)
# 向Local字典中添加一個名為stack的列表
# {pid:{"stack":[]}}
def push(self, obj):
rv = getattr(self._local, "stack", None)
if rv is None:
self._local.stack = rv = []
rv.append(obj)
return rv
# 消息閃現的實現原理，獲取或者移除
def pop(self):
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
release_local(self._local)
return stack[-1]
else:
return stack.pop()
# 只獲取
@property
def top(self):
try:
return self._local.stack[-1]
except (AttributeError, IndexError):
return None
LocalProxy
訪問Local時用LocalProxy，實際上是一個代理對象：
current_app = LocalProxy(_find_app)
request = LocalProxy(partial(_lookup_req_object, "request"))
session = LocalProxy(partial(_lookup_req_object, "session"))
g = LocalProxy(partial(_lookup_app_object, "g"))
這四句話代表四個意思，使用最多的范圍如下：
from flask import request
from flask import session
from flask import g
from flask import current_app
源碼如下：
@implements_bool
class LocalProxy(object):
__slots__ = ("__local", "__dict__", "__name__", "__wrapped__")
def __init__(self, local, name=None):
object.__setattr__(self, "_LocalProxy__local", local)
object.__setattr__(self, "__name__", name)
if callable(local) and not hasattr(local, "__release_local__"):
object.__setattr__(self, "__wrapped__", local)
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
return self.__local()
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
@property
def __dict__(self):
try:
return self._get_current_object().__dict__
except RuntimeError:
raise AttributeError("__dict__")
def __repr__(self):
try:
obj = self._get_current_object()
except RuntimeError:
return "<%s unbound>" % self.__class__.__name__
return repr(obj)
def __bool__(self):
try:
return bool(self._get_current_object())
except RuntimeError:
return False
def __unicode__(self):
try:
return unicode(self._get_current_object()) # noqa
except RuntimeError:
return repr(self)
def __dir__(self):
try:
return dir(self._get_current_object())
except RuntimeError:
return []
def __getattr__(self, name):
if name == "__members__":
return dir(self._get_current_object())
return getattr(self._get_current_object(), name)
def __setitem__(self, key, value):
self._get_current_object()[key] = value
def __delitem__(self, key):
del self._get_current_object()[key]
if PY2:
__getslice__ = lambda x, i, j: x._get_current_object()[i:j]
def __setslice__(self, i, j, seq):
self._get_current_object()[i:j] = seq
def __delslice__(self, i, j):
del self._get_current_object()[i:j]
__setattr__ = lambda x, n, v: setattr(x._get_current_object(), n, v)
__delattr__ = lambda x, n: delattr(x._get_current_object(), n)
__str__ = lambda x: str(x._get_current_object())
__lt__ = lambda x, o: x._get_current_object() < o
__le__ = lambda x, o: x._get_current_object() <= o
__eq__ = lambda x, o: x._get_current_object() == o
__ne__ = lambda x, o: x._get_current_object() != o
__gt__ = lambda x, o: x._get_current_object() > o
__ge__ = lambda x, o: x._get_current_object() >= o
__cmp__ = lambda x, o: cmp(x._get_current_object(), o) # noqa
__hash__ = lambda x: hash(x._get_current_object())
__call__ = lambda x, *a, **kw: x._get_current_object()(*a, **kw)
__len__ = lambda x: len(x._get_current_object())
__getitem__ = lambda x, i: x._get_current_object()[i]
__iter__ = lambda x: iter(x._get_current_object())
__contains__ = lambda x, i: i in x._get_current_object()
__add__ = lambda x, o: x._get_current_object() + o
__sub__ = lambda x, o: x._get_current_object() - o
__mul__ = lambda x, o: x._get_current_object() * o
__floordiv__ = lambda x, o: x._get_current_object() // o
__mod__ = lambda x, o: x._get_current_object() % o
__divmod__ = lambda x, o: x._get_current_object().__divmod__(o)
__pow__ = lambda x, o: x._get_current_object() ** o
__lshift__ = lambda x, o: x._get_current_object() << o
__rshift__ = lambda x, o: x._get_current_object() >> o
__and__ = lambda x, o: x._get_current_object() & o
__xor__ = lambda x, o: x._get_current_object() ^ o
__or__ = lambda x, o: x._get_current_object() | o
__div__ = lambda x, o: x._get_current_object().__div__(o)
__truediv__ = lambda x, o: x._get_current_object().__truediv__(o)
__neg__ = lambda x: -(x._get_current_object())
__pos__ = lambda x: +(x._get_current_object())
__abs__ = lambda x: abs(x._get_current_object())
__invert__ = lambda x: ~(x._get_current_object())
__complex__ = lambda x: complex(x._get_current_object())
__int__ = lambda x: int(x._get_current_object())
__long__ = lambda x: long(x._get_current_object()) # noqa
__float__ = lambda x: float(x._get_current_object())
__oct__ = lambda x: oct(x._get_current_object())
__hex__ = lambda x: hex(x._get_current_object())
__index__ = lambda x: x._get_current_object().__index__()
__coerce__ = lambda x, o: x._get_current_object().__coerce__(x, o)
__enter__ = lambda x: x._get_current_object().__enter__()
__exit__ = lambda x, *a, **kw: x._get_current_object().__exit__(*a, **kw)
__radd__ = lambda x, o: o + x._get_current_object()
__rsub__ = lambda x, o: o - x._get_current_object()
__rmul__ = lambda x, o: o * x._get_current_object()
__rdiv__ = lambda x, o: o / x._get_current_object()
if PY2:
__rtruediv__ = lambda x, o: x._get_current_object().__rtruediv__(o)
else:
__rtruediv__ = __rdiv__
__rfloordiv__ = lambda x, o: o // x._get_current_object()
__rmod__ = lambda x, o: o % x._get_current_object()
__rdivmod__ = lambda x, o: x._get_current_object().__rdivmod__(o)
__copy__ = lambda x: copy.copy(x._get_current_object())
__deepcopy__ = lambda x, memo: copy.deepcopy(x._get_current_object(), memo)
基本概念
在Flask中，每一次HTTP請求的到來都會執行一些操作。
舉個例子，Django里面request是通過形參的方式傳遞進視圖函數，這個很好實現，那么Flask中的request則是通過導入的方式作用於視圖函數，這意味着每次request中的數據都要進行更新。
它是如何做到的呢？這個就是Flask的精髓，上下文管理。
上面說過，Local對象會實例化兩次：
_app_ctx_stack = LocalStack()
_request_ctx_stack = LocalStack()
它的實現原理是這樣的，每一次HTTP請求來臨都會創建一個線程，Local對象就會依照這個線程的pid來構建出一個字典，這里用掉的對象是_request_ctx_stack，它內部有一個叫做__storage__的變量，最終會搞成下面的數據格式：
{
pid001:{stack:[<app_ctx = RequestContext request,session]}, # 存儲request對象以及session
pid002:{stack:[<app_ctx = RequestContext request,session]},
}
而除開_request_ctx_stack外還會有一個叫做_app_ctx_stack的東西，它會存放當前Flask實例app以及一個g對象：
{
pid001:{stack:[<app_ctx = flask.ctx.AppContext app,g>]},
pid002:{stack:[<app_ctx = flask.ctx.AppContext app,g>]},
}
每一次請求來的時候都會創建這樣的兩個字典，請求走的時候進行銷毀。
在每次導入request/session時都會從上面的這個__storage__字典中，拿出當前線程對應的pid中的request/session，以達到更行的目的。
| 類 | 功能 |
|---|---|
| Local | 構建大字典 |
| Localstack | 構建stack這個列表實現的棧 |
| LocalProxy | 控制獲取stack列表中棧的數據，如導入時引入request，怎么樣將stack中的request拿出來 |
Flask流程之__call__
Flask基本請求流程是建立在werkzeug之上：
from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple
@Request.application
def index(request):
return Response("Hello werkzeug")
if __name__ == '__main__':
run_simple("localhost", 5000, index)
可以看到，werkzeug在開啟服務后，會執行一個叫run_simple的函數，並且會調用被裝飾器包裝過后的index函數。
也就意味着，在run_simple傳參時，第三個參數形參名application會加括號進行調用。
如果你傳入一個類，它將執行__init__方法，如果你傳入一個實例對象，它將執行其類的__call__方法。
如下所示：
from werkzeug.serving import run_simple
class Test:
def __init__(self,*args,**kwargs):
print("run init")
if __name__ == '__main__':
run_simple("localhost", 5000, Test)
# run init
示例二：
from werkzeug.serving import run_simple
class Test:
def __init__(self,*args,**kwargs):
super(Test, self).__init__(*args,**kwargs)
def __call__(self, *args, **kwargs):
print("run call")
test = Test()
if __name__ == '__main__':
run_simple("localhost", 5000, test)
# run call
OK，現在牢記一點，如果傳入的是函數，執行其類的__call__。
接下來我們看Flask程序：
from flask import Flask
app = Flask(__name__)
@app.route('/index')
def index():
return "index"
if __name__ == '__main__':
app.run()
當請求來時，會執行run方法，我們朝里看源碼，直接拉到run方法的下面，在run方法中調用了run_simple方法，其中的第三個參數就是當前的Flask實例對象app：
try:
run_simple(host, port, self, **options)
finally:
self._got_first_request = False
所以到了這里，實例調用父類的__call__（Flask類本身），繼續看app.__call__，實例本身沒有找其類的。
# app.__call__ 鼠標左鍵點進去
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response)
可以看見，在這里它調用的是app.wsgi_app方法，這也就能解釋Flask中間件為什么重寫下面這段代碼。
app.wsgi_app = Middleware(app.wsgi_app) # 傳入要原本執行的wsgi_app
Flask流程之wsgi_app
原生的Flask.wsgi_app中的代碼是整個Flask框架中的核心，如下所示：
def wsgi_app(self, environ, start_response):
# 參數 self就是Flask實例對象app，environ是werkzeug的原生HTTP請求對象
ctx = self.request_context(environ) # 對environ這個原生的請求對象進行封裝，封裝成一個request對象，可以擁有request.method/.args/.form等方法
error = None
try:
try:
# 上下文管理，將ctx放入Local大字典中，並且會更新session，從HTTP請求中拿出session，此外還會將當前實例app，也就是self放入Local的另一個大字典中，當然還有g對象
ctx.push()
# 執行視圖函數
response = self.full_dispatch_request()
# 捕獲異常
except Exception as e:
error = e
response = self.handle_exception(e)
except:
error = sys.exc_info()[1]
raise
# 返回請求
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
# 清除上下文管理內容
ctx.auto_pop(error)
Flask流程之Resquest
看下面這一行代碼，它其實是實例化一個對象，用於封裝environ這個原始的HTTP請求：
def wsgi_app(self, environ, start_response):
# self：app，也就是Flask實例
ctx = self.request_context(environ)
點開它，發現會返回一個實例對象：
def request_context(self, environ):
# self：app，也就是Flask實例
return RequestContext(self, environ) # 注意這里傳參，__init__的第二個參數是app實例
去找它的__init__方法：
class RequestContext(object):
def __init__(self, app, environ, request=None, session=None):
# self：ReuqetContext對象 app是Flask實例
self.app = app
if request is None:
request = app.request_class(environ) # 執行這里、封裝request
self.request = request
self.url_adapter = None
try:
self.url_adapter = app.create_url_adapter(self.request) # 創建url與視圖函數對應關系，這里不看
except HTTPException as e:
self.request.routing_exception = e
self.flashes = None
self.session = session # 注意 session 是一個None
先看上面的一句，封裝request對象，由於self是app，所以找Flask類中的request_class，它是一個類屬性：
request_class = Request
加括號進行調用，並且傳遞了environ，所以要進行實例化，找它的__init__方法，發現是一個多繼承類：
class Request(RequestBase, JSONMixin):
# 該類本身未實現__init__，去找它的父類，從左到右找
找它的父類，RequestBase，也是一個多繼承類：
class Request(
BaseRequest,
AcceptMixin,
ETagRequestMixin,
UserAgentMixin,
AuthorizationMixin,
CORSRequestMixin,
CommonRequestDescriptorsMixin,
):
# 該類本身未實現__init__，去找它的父類，從上到下
再繼續向上找，找BaseRequest類：
class BaseRequest(object):
def __init__(self, environ, populate_request=True, shallow=False):
# self：Request，因為是Request對象要實例化
self.environ = environ
if populate_request and not shallow:
self.environ["werkzeug.request"] = self
self.shallow = shallow
# 該類還實現了args\form\files等方法
# 大體意思就是說調用這些方法的時候會將environ中的數據解析出來
然后將結果返回給ctx，這個ctx就是RequestContext的實例對象，里面有個Request實例對象request，還有個session，不過是None：
<ctx=RequestContext request,session=None>
Flask流程之ctx.push
接着往下看代碼，記住現在的線索：
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ)
# <ctx=RequestContext request,session=None>
error = None
try:
try:
ctx.push() # 接下來着重看這里
response = self.full_dispatch_request()
except Exception as e:
執行ctx.push，這個方法可以說非常的繞。
def push(self):
# 參數：self是ctx，也就是 <ctx=RequestContext request,session=None>
# _request_ctx_stack = LocalStack() 全局變量，已經做好了
top = _request_ctx_stack.top # None
if top is not None and top.preserved: # False 不走這里
top.pop(top._preserved_exc)
# _app_ctx_stack = LocalStack() 全局變量，已經做好了
app_ctx = _app_ctx_stack.top # None
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)
if self.session is None:
session_interface = self.app.session_interface
self.session = session_interface.open_session(self.app, self.request)
if self.session is None:
self.session = session_interface.make_null_session(self.app)
if self.url_adapter is not None:
self.match_request()
現在來看 app_ctx到底是個神馬玩意兒：
def app_context(self):
# self：Flask實例化對象，app
return AppContext(self)
繼續走：
class AppContext(object):
def __init__(self, app):
# self：AppContext的實例對象，app就是Flask的實例對象
self.app = app # AppContext實例對象的app就是Flask的實例對象
self.url_adapter = app.create_url_adapter(None)
self.g = app.app_ctx_globals_class() # 一個空的g對象
self._refcnt = 0
現在回來，第二個重要點來了：
app_ctx = _app_ctx_stack.top # None
if app_ctx is None or app_ctx.app != self.app: # 會走這里，因為第一個條件成立
app_ctx = self.app.app_context() # <app_ctx = flask.ctx.AppContext app,g> app是當前Flask實例對象，g是一個空的玩意兒，詳細代碼不用看了
app_ctx.push() # 又執行push
self._implicit_app_ctx_stack.append(app_ctx)
else:
self._implicit_app_ctx_stack.append(None)
Flask流程之app_ctx.push
這個push是AppContext中的push：
def push(self):
# self： <app_ctx = flask.ctx.AppContext app,g>
self._refcnt += 1
if hasattr(sys, "exc_clear"):
sys.exc_clear()
_app_ctx_stack.push(self) # 存入
appcontext_pushed.send(self.app) # 這里不看了
其實就是往Local的大字典中進行存放，不過是通過LocalStack這個類的push方法：
def push(self, obj):
rv = getattr(self._local, "stack", None) # 觸發Local.__getattr__返回一個None
if rv is None: # 走這里，觸發Local.__setattr__
self._local.stack = rv = []
rv.append(obj) # 直接存,觸發Local.__setattr__
return rv
數據結構：
{
pid001:{stack:[<app_ctx = flask.ctx.AppContext app,g>]},
}
然后返回app_ctx.push:
app_ctx = _app_ctx_stack.top # None
if app_ctx is None or app_ctx.app != self.app: # 會走這里，因為第一個條件成立
app_ctx = self.app.app_context() # <app_ctx = flask.ctx.AppContext app,g> app是當前Flask實例對象，g是一個空的玩意兒，詳細代碼不用看了
app_ctx.push() # 存入成功
self._implicit_app_ctx_stack.append(app_ctx) # 走這里了 ctx類RequestContext中有一個列表，把他存進來
else:
self._implicit_app_ctx_stack.append(None)
Flask流程之_request_ctx_stack.push
繼續向下看ctx.push中的代碼，這里主要是封裝請求上下文：
def push(self):
# self：ctx對象 <ctx=RequestContext request,session=None>
# _request_ctx_stack = LocalStack() 全局變量，已經做好了
top = _request_ctx_stack.top # None
if top is not None and top.preserved: # False 不走這里
top.pop(top._preserved_exc)
# _app_ctx_stack = LocalStack() 全局變量，已經做好了
app_ctx = _app_ctx_stack.top # None
if app_ctx is None or app_ctx.app != self.app: # 會走這里，因為第一個條件成立
app_ctx = self.app.app_context() # 走這里，實際上就是實例化
app_ctx.push() # 存入Local中
self._implicit_app_ctx_stack.append(app_ctx) # 存入ctx的列表中
else:
self._implicit_app_ctx_stack.append(None)
if hasattr(sys, "exc_clear"): # 不管
sys.exc_clear()
_request_ctx_stack.push(self) # 重點是這里，和上面相同，又創建了一個字典。放進去,需要注意的是這里是_request_ctx_stack.push,是另一個不同的Local實例化對象
"""
{
pid001:{stack:[<app_ctx = RequestContext request,session=None]},
}
"""
if self.session is None: # 設置session
session_interface = self.app.session_interface
self.session = session_interface.open_session(self.app, self.request) # 打開session，從request中讀取出cookie然后進行load反序列化
if self.session is None:
self.session = session_interface.make_null_session(self.app)
"""
現在session就不是None了
{
pid001:{stack:[<app_ctx = RequestContext request,session]},
}
"""
if self.url_adapter is not None:
self.match_request()
Flask流程之session
保存session，從Cookie中獲取數據，反序列化后保存到session中。
class SecureCookieSessionInterface(SessionInterface):
serializer = session_json_serializer
session_class = SecureCookieSession
def get_signing_serializer(self, app):
if not app.secret_key:
return None
signer_kwargs = dict(
key_derivation=self.key_derivation, digest_method=self.digest_method
)
return URLSafeTimedSerializer(
app.secret_key,
salt=self.salt,
serializer=self.serializer,
signer_kwargs=signer_kwargs,
)
def open_session(self, app, request):
s = self.get_signing_serializer(app)
if s is None:
return None
val = request.cookies.get(app.session_cookie_name)
if not val:
return self.session_class()
max_age = total_seconds(app.permanent_session_lifetime)
try:
data = s.loads(val, max_age=max_age)
return self.session_class(data)
except BadSignature:
return self.session_class()
def save_session(self, app, session, response):
domain = self.get_cookie_domain(app)
path = self.get_cookie_path(app)
# If the session is modified to be empty, remove the cookie.
# If the session is empty, return without setting the cookie.
if not session:
if session.modified:
response.delete_cookie(
app.session_cookie_name, domain=domain, path=path
)
return
# Add a "Vary: Cookie" header if the session was accessed at all.
if session.accessed:
response.vary.add("Cookie")
if not self.should_set_cookie(app, session):
return
httponly = self.get_cookie_httponly(app)
secure = self.get_cookie_secure(app)
samesite = self.get_cookie_samesite(app)
expires = self.get_expiration_time(app, session)
val = self.get_signing_serializer(app).dumps(dict(session))
response.set_cookie(
app.session_cookie_name,
val,
expires=expires,
httponly=httponly,
domain=domain,
path=path,
secure=secure,
samesite=samesite,
)
導入源碼
如果使用from flask import request它會通過LocalProxy去拿到Local中的ctx對象然后進行解析，拿到request對象。
request = LocalProxy(partial(_lookup_req_object, "request")) # 實例化
在這里可以看見實例化了一個LocalProxy對象：
@implements_bool
class LocalProxy(object):
def __init__(self, local, name=None):
# local：偏函數
# name：None
object.__setattr__(self, "_LocalProxy__local", local) # self.__local = local 雙下開頭會改變名字
object.__setattr__(self, "__name__", name) # None
if callable(local) and not hasattr(local, "__release_local__"): # 運行，如果偏函數可執行，並且偏函數沒有屬性__release_local__時執行
object.__setattr__(self, "__wrapped__", local) # 當前實例增加屬性，指向偏函數
當要使用request.method時，觸發LocalProxy的__getattr__方法：
def __getattr__(self, name):
# name:method
if name == "__members__":
return dir(self._get_current_object())
return getattr(self._get_current_object(), name) # 執行這里，name = method
在_get_current_object中執行self.__local()：
def _get_current_object(self):
if not hasattr(self.__local, "__release_local__"):
return self.__local() # self.__local就是偏函數，偏函數自動傳參。request
try:
return getattr(self.__local, self.__name__)
except AttributeError:
raise RuntimeError("no object bound to %s" % self.__name__)
偏函數，第二個參數是request，也就是說_lookup_req_object的參數默認就是request：
def _lookup_req_object(name):
# name：request字符串
top = _request_ctx_stack.top # 返回RequestContext對象，里面有request和session
if top is None:
raise RuntimeError(_request_ctx_err_msg)
return getattr(top, name) # 返回的就是request對象，從RequestContext對象中拿到request對象
拿到request對象后繼續看__getattr__方法，獲取method：
def __getattr__(self, name):
# name:method
if name == "__members__":
return dir(self._get_current_object())
return getattr(self._get_current_object(), name) # 執行這里，name = method
Flask流程之pop
wsgi_app返回和清棧還沒有看：
def wsgi_app(self, environ, start_response):
ctx = self.request_context(environ) # 封裝request、session到RequestContext對象
error = None
try:
try:
ctx.push() # 封裝app上下文和請求上下文，填充session內容
response = self.full_dispatch_request() # 執行視圖函數
except Exception as e:
error = e
response = self.handle_exception(e)
except: # noqa: B001
error = sys.exc_info()[1]
raise
return response(environ, start_response) # 封裝返回對象
finally:
if self.should_ignore_error(error): # 清除上下文，兩個字典中的內容
error = None
ctx.auto_pop(error)
最主要就是看清棧：
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) # 看這里就行了
接着看pop方法：
def pop(self, exc=_sentinel):
app_ctx = self._implicit_app_ctx_stack.pop() # 清除Flask實例中存放的app和g對象， [<app_ctx = flask.ctx.AppContext app,g>]
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]
self.app.do_teardown_request(exc)
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() # # LocalProxy.pop()
# 修改request中的werkzeug.request為None，本身是Request對象本身
if clear_request:
rv.request.environ["werkzeug.request"] = None
# 清除應用上下文中的數據
if app_ctx is not None:
app_ctx.pop(exc) # LocalProxy.pop()
assert rv is self, "Popped wrong request context. (%r instead of %r)" % (
rv,
self,
)
LocalProxy.pop中的代碼：
def pop(self):
stack = getattr(self._local, "stack", None)
if stack is None:
return None
elif len(stack) == 1:
release_local(self._local) # Local.__storage__.pop(self.__ident_func__(), None)
return stack[-1] # 如果只剩下一個，就返回
else:
return stack.pop() # 清理干凈 {pid:{"statck":[]}}
Flask流程之before_request
before_request實現挺簡單的，用一個列表，將所有被裝飾函數放進來。再執行視圖函數之前把列表中所有被berore_request裝飾的函數先執行一遍：
@setupmethod
def before_request(self, f):
self：app，Flask實例
f：被裝飾函數
self.before_request_funcs.setdefault(None, []).append(f) # 從與i個字典中獲取None，如果沒獲取到就是一個空列表，后面使用了append代表None對應的k就是一個列表。
# 這句話的意思就是說，從一個字典中{None:[func,func,func]}出一個列表，獲取不到就創建一個空列表{None:[]}，並且把被裝飾的函數f添加進去
return f
在wsgi_app中查看源碼：
# 查看執行視圖函數這一句
response = self.full_dispatch_request()
點進去看，執行視圖函數前發生了什么：
def full_dispatch_request(self):
self.try_trigger_before_first_request_functions() # 先執行這berfor_first_request裝飾的函數
try:
request_started.send(self)
rv = self.preprocess_request() # 在執行berfor_first_request裝飾的函數
if rv is None:
rv = self.dispatch_request() # 開始執行視圖函數，rv=返回值
except Exception as e:
rv = self.handle_user_exception(e)
return self.finalize_request(rv)
關鍵代碼：
def preprocess_request(self):
# self：app，Flask實例對象
bp = _request_ctx_stack.top.request.blueprint # 獲取藍圖
funcs = self.url_value_preprocessors.get(None, ())
if bp is not None and bp in self.url_value_preprocessors:
funcs = chain(funcs, self.url_value_preprocessors[bp])
for func in funcs:
func(request.endpoint, request.view_args)
funcs = self.before_request_funcs.get(None, ()) # 獲取列表，[func1,func2]，key是None
if bp is not None and bp in self.before_request_funcs: # 如果藍圖存在，將藍圖的全局before_request也添加到列表中
funcs = chain(funcs, self.before_request_funcs[bp])
for func in funcs: # 運河，執行
rv = func()
if rv is not None: # 返回值如果不是None就攔截
return rv
而使用after_request裝上后的函數也會被放到一個列表中，其他的具體實現都差不多：
@setupmethod
def after_request(self, f):
self：app，Flask實例
f：被裝飾函數
self.after_request_funcs.setdefault(None, []).append(f)
return f
源碼流程圖
g的作用
一次請求流程中的一些共同數據，可以用g進行存儲：
from flask import Flask,g
app = Flask(__name__)
@app.before_request
def func():
g.message = "僅當次請求有效"
@app.route('/index')
def index():
print(g.message)
return "index"
if __name__ == '__main__':
app.run()
current_app的作用
可以導入當前的配置：
from flask import Flask,current_app
app = Flask(__name__)
@app.before_request
def func():
if current_app.debug == False:
current_app.debug = True
print("修改成功")
@app.route('/index')
def index():
return "index"
if __name__ == '__main__':
app.run()
WTforms
wtforms類似於Django中的forms組件，用於數據驗證和生成HTML
官方文檔：https://wtforms.readthedocs.io/en/stable/index.html#
基本使用
首先進行安裝：
pip3 install wtforms
一個簡單的注冊示例：
from flask import Flask
from flask import request
from flask import render_template
from flask import Markup
from wtforms import Form # 必須繼承
from wtforms import validators # 自定義認證器
from wtforms import widgets # HTML生成插件
from wtforms import fields # 字段導入
class LoginForm(Form):
name = fields.StringField(
label="用戶名",
widget=widgets.TextInput(),
render_kw={"class": "form-control"},
validators=[
validators.DataRequired(message="用戶名不能為空"),
validators.Length(max=8, min=3, message="用戶名長度必須大於%(max)d且小於%(min)d")
]
)
pwd = fields.PasswordField(
label="密碼",
widget=widgets.PasswordInput(),
render_kw={"class": "form-control",},
validators=[
validators.DataRequired(message="密碼不能為空"),
validators.Length(max=18, min=4, message="密碼長度必須大於%(max)d且小於%(min)d"),
validators.Regexp(regex="\d+", message="密碼必須是數字"),
]
)
app = Flask(__name__,template_folder="templates")
@app.route('/login',methods=["GET","POST"])
def login():
if request.method == "GET":
form = LoginForm()
return render_template("login.html",**{"form":form})
form = LoginForm(formdata=request.form)
if form.validate():
print("用戶提交的數據用過格式驗證，值為：%s" % form.data)
return "登錄成功"
else:
print(form.errors, "錯誤信息")
return render_template("login.html", **{"form":form})
if __name__ == '__main__':
app.run()
前端渲染：
<form method="POST" novalidate>
{% for item in form %}
<p> {{item.label}}:{{item}}</p>
<p style="color: red">{{item.errors[0]}}</p>
{% endfor %}
<p>
<button type="submit">提交</button>
</p>
</form>
Form實例化
以下是Form類的實例化參數：
| 參數 | 描述 |
|---|---|
| formdata | 需要被驗證的form表單數據 |
| obj | 如果formdata為空或未提供，則檢查此對象的屬性是否與表單字段名稱匹配，這些屬性將用於字段值 |
| prefix | 字段前綴匹配，當傳入該參數時，所有驗證字段必須以這個開頭(無太大意義) |
| data | 當formdata參數和obj參數都有時候，可以使用該參數傳入字典格式的待驗證數據或者生成html的默認值，列如：{'usernam':'admin’} |
| meta | 用於覆蓋當前已經定義的form類的meta配置，參數格式為字典 |
使用data來構建默認值，常用於文章編輯等，需要填入原本數據庫中查詢出的文章。
下面用默認用戶做演示：
def login():
if request.method == "GET":
form = LoginForm(data={"name":"默認用戶"})
return render_template("login.html",**{"form":form})
字段介紹
字段的繼承，fields是常用類，它繼承了其他的一些類：
from wtforms.fields.core import *
from wtforms.fields.simple import *
from wtforms.fields.core import Label, Field, SelectFieldBase, Flags
from wtforms.utils import unset_value as _unset_value
一般都字段都會默認生成一種HTML標簽，但是也可以通過widget進行更改，下面是常用的一些字段：
| 字段類型 | 描述 |
|---|---|
| StringField | 文本字段， 相當於type類型為text的input標簽 |
| TextAreaField | 多行文本字段 |
| PasswordField | 密碼文本字段 |
| HiddenField | 隱藏文本字段 |
| DateField | 文本字段， 值為datetime.date格式 |
| DateTimeField | 文本字段， 值為datetime.datetime格式 |
| IntegerField | 文本字段， 值為整數 |
| DecimalField | 文本字段， 值為decimal.Decimal |
| FloatField | 文本字段， 值為浮點數 |
| BooleanField | 復選框， 值為True 和 False |
| RadioField | 一組單選框 |
| SelectField | 下拉列表 |
| SelectMultipleField | 下拉列表， 可選擇多個值 |
| FileField | 文件上傳字段 |
| SubmitField | 表單提交按鈕 |
| FormFiled | 把表單作為字段嵌入另一個表單 |
| FieldList | 子組指定類型的字段 |
每個字段可以為其配置一些額外的屬性，如下所示：
| 字段屬性 | 描述 |
|---|---|
| label | 字段別名 |
| validators | 驗證規則列表 |
| filters | 過濾器列表 |
| description | 字段詳細描述 |
| default | 默認值 |
| widget | 自定義插件，替換默認生成的HTML標簽 |
| render_kw | 為生成的HTML標簽配置屬性 |
| choices | 復選框的類型 |
如下所示：
class FormLearn(Form):
field = fields.StringField(
label="測試字段",
widget=widgets.TextInput(), # 使用插件，代替默認生成標簽
validators=[
validators.DataRequired(message="必填"),
validators.Length(max=12,min=3,message="長度驗證"),
],
description="這是一個測試字段",
default="默認值",
render_kw={"style":"width:60px"}, # 設置鍵值對
)
內置驗證
使用validators為字段進行驗證時，可指定如下的內置驗證規則：
| 驗證函數 | 說明 |
|---|---|
| 驗證電子郵件地址 | |
| EqualTo | 比較兩個字段的值，常用於要求輸入兩次密碼進行確認的情況 |
| IPAddress | 驗證IPv4網絡地址 |
| Length | 驗證輸入字符串的長度 |
| NumberRange | 驗證輸入的值在數字范圍內 |
| Optional | 無輸入值時跳過其他驗證函數 |
| DataRequired | 確保字段中有數據 |
| Regexp | 使用正則表達式驗證輸入值 |
| URL | 驗證URL |
| AnyOf | 確保輸入值在可選值列表中 |
| NoneOf | 確保輸入值不在可選列表中 |
EqualTo是常用的驗證方式，驗證兩次密碼是否輸入一致：
class FormLearn(Form):
password = fields.StringField(
label="用戶密碼",
widget=widgets.PasswordInput(),
validators=[
validators.DataRequired(message="必填"),
validators.Length(min=8, max=16, message="必須小於8位大於16位")
],
)
re_password = fields.StringField(
label="密碼驗證",
widget=widgets.PasswordInput(),
validators=[
validators.EqualTo(fieldname="password", message="兩次密碼輸入不一致", )
],
render_kw={"placeholder": "重新輸入密碼"},
)
Meta配置
Meta主要用於自定義wtforms的功能，用的比較少，大多都是配置選項，以下是配置參數：
from wtforms import Form # 必須繼承
from wtforms import fields # 字段導入
from wtforms.csrf.core import CSRF # 自帶的CSRF驗證和生成
from hashlib import md5 # 加密
class MyCSRF(CSRF):
def setup_form(self, form):
self.csrf_context = form.meta.csrf_context()
self.csrf_secret = form.meta.csrf_secret
return super(MyCSRF, self).setup_form(form)
def generate_csrf_token(self, csrf_token):
gid = self.csrf_secret + self.csrf_context
token = md5(gid.encode('utf-8')).hexdigest()
return token
def validate_csrf_token(self, form, field):
if field.data != field.current_token:
raise ValueError('Invalid CSRF')
class FormLearn(Form):
username = fields.StringField(label="用戶名")
password = fields.PasswordField(label="密碼")
class Meta:
# CSRF相關
csrf = False # 是否自動生成CSRF標簽
csrf_field_name = "csrf_token" # 生成的CSRF標簽名字
csrf_secret = "2d728321*fd&" # 自動生成標簽的值，加密用csrf_context
csrf_context = lambda x:request.url
csrf_class = MyCSRF # 生成和比較的CSRF標簽
# 其他配置
locales = ('zh', 'en') # 是否支持本地化 locales = False
cache_translations = True # 是否對本地化進行緩存
translations_cache = {} # 保存本地化緩存信息的字段
鈎子函數
一般都用局部鈎子：
class FormLearn(Form):
username = fields.StringField(label="用戶名")
password = fields.PasswordField(label="密碼")
def validate_username(self,obj):
"""
:param obj: 字段對象，屬性data就是用戶輸入的內容
:return: 如果不進行返回，則默認返回obj對象
"""
if len(obj.data) < 6:
# raise validators.ValidationError("用戶名太短了") # 繼續后續驗證
raise validators.StopValidation("用戶名太短了") # 不再繼續后續驗證
return obj
def validate_password(self,obj):
print("局部鈎子")
return obj
自定義驗證
自定義驗證規則：
from wtforms import validators
class ValidatorsRule(object):
"""自定義驗證規則"""
def __call__(self, form, field):
"""
:param form:
:param field: 使用field.data，取出用戶輸入的信息
:return: 當return是None，則驗證通過
"""
import re
error_char = re.search(r"\W", field.data).group(0) # 取出第一個匹配結果
if error_char:
raise validators.StopValidation("提交數據含有特殊字符,如:%s"%error_char") # 不再繼續后續驗證
# raise validators.ValidationError("提交數據含有特殊字符,如:%s"%error_char) # 繼續后續驗證
使用：
class LoginForm(Form):
name = simple.StringField(
label="用戶名",
widget=widgets.TextInput(),
render_kw={"class": "form-control"},
validators=[
validators.DataRequired(message="用戶名不能為空"),
validators.Length(max=8, min=3, message="用戶名長度必須大於%(max)d且小於%(min)d"),
ValidatorsRule(), # 使用自定義驗證
]
)
插件大全
以下代碼包含所有可能用到的插件：
from flask import Flask,render_template,redirect,request
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
=======================simple===========================
class RegisterForm(Form):
name = simple.StringField(
label="用戶名",
validators=[
validators.DataRequired()
],
widget=widgets.TextInput(),
render_kw={"class":"form-control"},
default="wd"
)
pwd = simple.PasswordField(
label="密碼",
validators=[
validators.DataRequired(message="密碼不能為空")
]
)
pwd_confim = simple.PasswordField(
label="重復密碼",
validators=[
validators.DataRequired(message='重復密碼不能為空.'),
validators.EqualTo('pwd',message="兩次密碼不一致")
],
widget=widgets.PasswordInput(),
render_kw={'class': 'form-control'}
)
========================html5============================
email = html5.EmailField( #注意這里用的是html5.EmailField
label='郵箱',
validators=[
validators.DataRequired(message='郵箱不能為空.'),
validators.Email(message='郵箱格式錯誤')
],
widget=widgets.TextInput(input_type='email'),
render_kw={'class': 'form-control'}
)
===================以下是用core來調用的=======================
gender = core.RadioField(
label="性別",
choices=(
(1,"男"),
(1,"女"),
),
coerce=int # 傳入時自動轉換位int類型，否則是str類型
)
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): #這里的self是一個RegisterForm對象
'''
解決數據庫不及時更新的問題
重寫__init__方法
'''
super(RegisterForm,self).__init__(*args, **kwargs) #繼承父類的init方法
self.favor.choices =((1, '籃球'), (2, '足球'), (3, '羽毛球')) #把RegisterForm這個類里面的favor重新賦值，實現動態改變復選框中的選項
def validate_pwd_confim(self,field,):
'''
自定義pwd_config字段規則，例：與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': 1}) #默認是1,
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()
源碼解析
Flask-session
、通過第三方插件Flask-session能夠將session存放至其他地方，而不是只能存放在內存中：
pip install Flask-session
使用的時候：
from flask import Flask
from flask import session
from flask_session import Session
import redis
app = Flask(__name__)
app.config["SESSION_TYPE"] = "redis"
app.config["SESSION_REDIS"] = redis.Redis(host="127.0.0.1",port=6379,password="") # 有密碼就填上
app.config["SESSION_KEY_PREFIX"] = "session" # 前綴
Session(app) # 修改flask的默認session接口
@app.route('/set')
def set():
session["key"] = "value"
return "ok"
@app.route('/get')
def get():
res = session.get("key")
return res
if __name__ == '__main__':
app.run()
原理也很簡單，替換掉了默認的session接口，與Django的redis緩存差不多。
Flask-SQLALchemy
Flask-SQLALchemy是SQLALchemy與Flask之間的粘合劑。讓Flask與SQLALchemy之間的關系更為緊密：
pip install flask-sqlalchemy
使用Flask-SQLALchemy也非常簡單，首先是創建項目：
- mysite # 項目根目錄
- mysite # 包
- static
- templates
- views
index.py
- __init__.py
- models.py
- manage.py
- settings.py
代碼如下，做主藍圖，使用flaskSQLAlchemy模塊：
# __init__.py
from flask import Flask
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy
# 第一步，實例化
db = SQLAlchemy() # db包含了所有需要的東西，如commit，remove，Base類等
from .models import * # 導入模型類
from .views.index import index # 防止循環導入
def create_app():
app = Flask(import_name=__name__, template_folder='../templates', static_folder='../static',
static_url_path='/static')
# 加載配置文件
app.config.from_pyfile("../settings.py")
# 將db注冊到app中，必須在注冊藍圖之前
db.init_app(app)
# 配置Session
Session(app)
# 注冊藍圖
app.register_blueprint(index, url_prefix="/index/") # 注冊藍圖對象 index
return app
settings.py中的配置項：
import redis
# sqlalchemy相關配置
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root@127.0.0.1:3306/db2?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5 # 鏈接池的連接數量
SQLALCHEMY_POOL_TIMEOUT = 10 # 鏈接池連接超時時間
SQLALCHEMY_POOL_RECYCLE = 60*60*4 # 關閉連接的時間：默認Mysql是2小時
SQLALCHEMY_MAX_OVERFLOW = 3 # 控制在連接池達到最大值后可以創建的連接數。當這些額外的連接回收到連接池后將會被斷開和拋棄
SQLALCHEMY_TRACK_MODIFICATIONS = False # 追蹤對象的修改並且發送信號
# session相關配置
SESSION_TYPE = 'redis' # session類型為redis
SESSION_KEY_PREFIX = 'session:' # 保存到session中的值的前綴
SESSION_PERMANENT = True # 如果設置為False，則關閉瀏覽器session就失效。
SESSION_USE_SIGNER = False # 是否對發送到瀏覽器上 session:cookie值進行加密
SESSION_REDIS = redis.Redis(host="127.0.0.1",port=6379,password="")
然后是書寫模型類：
# models.py
from . import db
class UserProfile(db.Model): # 必須繼承Base
__tablename__ = 'userprofile'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
password = db.Column(db.String(64), unique=True, nullable=False)
email = db.Column(db.String(128), unique=True, nullable=False)
def __repr__(self):
return '<user %s>' % self.username
視圖：
# index.py
from flask import Blueprint
from .. import db
from .. import models
index = Blueprint("index", __name__)
@index.route("/model")
def model():
import uuid
db.session.add( # 使用session添加數據
models.UserProfile(
username="user%s" % (uuid.uuid4()),
password="password%s" % str(uuid.uuid4()),
email="%s@gamil.com" % str(uuid.uuid4())
)
)
db.session.commit() # 提交
result = db.session.query(models.UserProfile).all() # 查詢數據
db.session.remove() # 關閉
print(result)
return "ok"
啟動文件：
# manage.py
from mysite import create_app
from mysite import db
if __name__ == '__main__':
app = create_app()
with app.app_context(): # 執行腳本創建數據庫表
# db.drop_all()
db.create_all()
app.run()
Flask-Script
該插件的作用通過腳本的形式啟動Flask項目，同時還具有自定義腳本命令的功能。
下載安裝：
pip install flask-script
在啟動文件中進行使用：
from flask_script import Manager
from mysite import create_app
from mysite import db
if __name__ == '__main__':
app = create_app()
manager = Manager(app) # 注冊
with app.app_context(): # 執行腳本創建數據庫表
# db.drop_all()
db.create_all()
manager.run() # 使用manager.run啟動flask項目
啟動命令：
python manage.py runserver -h 127.0.0.1 -p 5000
你可以自定義一些啟動腳本，如下所示：
@manager.command
def cmd(args):
print(args)
# 命令:python manage.py cmd 12345
# 結果:12345
也可以使用關鍵字傳參的方式，進行腳本的啟動：
@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):
print(name, url)
# 命令:python manage.py cmd -n test -u www.xxx.com
# 結果:test www.xxx.com
自定義腳本可以配置是否創建數據庫，配合Flask-SQLAlchemy，如下所示：
from flask_script import Manager
from mysite import create_app
from mysite import db
if __name__ == '__main__':
app = create_app()
manager = Manager(app) # 注冊
@manager.command
def create_tables():
with app.app_context(): # 執行腳本創建數據庫表
# db.drop_all()
db.create_all()
manager.run() # 使用manager.run啟動flask項目
# 命令:python manage.py create_tables
Flask-Migrate
該插件的作用類似於Django中對model的命令行操作，由於原生Flask-SQLALchemy不支持表結構的修改，所以用該插件的命令行來彌補。
值得一提的是，該插件依賴於Flask-Script：
pip install flask-migrate
在啟動文件中進行導入：
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from mysite import create_app
if __name__ == '__main__':
app = create_app()
manager = Manager(app) # 注冊 flask-scripts組件
Migrate(app) # 注冊 flask-migrate組件
manager.add_command("db",MigrateCommand)
manager.run() # 使用manager.run啟動flask項目
命令行：
| 命令 | 描述 |
|---|---|
| python 啟動文件.py db init | 初始化model |
| python 啟動文件.py db migrate | 類型於makemigrations，生成模型類 |
| python 啟動文件.py db upgrade | 類似於migrate，將模型類映射到物理表中 |
在第一次使用時，三條命令都敲一遍。
如果修改了表結構，只用敲第二條，第三條命令即可，彌補Flask-SQLALchemy不能修改表結構的缺點。
最后記錄
一個完整基礎的Flask項目基礎架構：
- mysite # 項目根目錄
- mysite # 包
- static
- templates
- views
index.py
- __init__.py
- models.py
- manage.py
- settings.py
# __init__.py
from flask import Flask
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy
# 第一步，實例化
db = SQLAlchemy()
from .models import *
from .views.index import index # 防止循環導入
def create_app():
app = Flask(import_name=__name__, template_folder='../templates', static_folder='../static',
static_url_path='/static')
# 加載配置文件
app.config.from_pyfile("../settings.py")
# 將db注冊到app中，必須在注冊藍圖之前
db.init_app(app)
# 配置Session
Session(app)
# 注冊藍圖
app.register_blueprint(index, url_prefix="/index/") # 注冊藍圖對象 index
return app
# models.py
from . import db
class UserProfile(db.Model):
__tablename__ = 'userprofile'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
password = db.Column(db.String(64), unique=True, nullable=False)
# email = db.Column(db.String(128), unique=True, nullable=False)
def __repr__(self):
return '<user %s>' % self.username
# manage.py
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from mysite import create_app
from mysite import db
if __name__ == '__main__':
app = create_app()
manager = Manager(app) # 注冊 flask-scripts組件
Migrate(app,db) # 注冊 flask-migrate組件
manager.add_command("db",MigrateCommand)
manager.run() # 使用manager.run啟動flask項目
# settings.py
import redis
# sqlalchemy相關配置
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root@127.0.0.1:3306/db2?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5 # 鏈接池的連接數量
SQLALCHEMY_POOL_TIMEOUT = 10 # 鏈接池連接超時時間
SQLALCHEMY_POOL_RECYCLE = 60*60*4 # 關閉連接的時間：默認Mysql是2小時
SQLALCHEMY_MAX_OVERFLOW = 3 # 控制在連接池達到最大值后可以創建的連接數。當這些額外的連接回收到連接池后將會被斷開和拋棄
SQLALCHEMY_TRACK_MODIFICATIONS = False # 追蹤對象的修改並且發送信號
# session相關配置
SESSION_TYPE = 'redis' # session類型為redis
SESSION_KEY_PREFIX = 'session:' # 保存到session中的值的前綴
SESSION_PERMANENT = True # 如果設置為False，則關閉瀏覽器session就失效。
SESSION_USE_SIGNER = False # 是否對發送到瀏覽器上 session:cookie值進行加密
SESSION_REDIS = redis.Redis(host="127.0.0.1",port=6379,password="")
# index.py
from flask import Blueprint
from .. import db
from .. import models
index = Blueprint("index", __name__)
@index.route("/model")
def model():
# 插入數據
import uuid
db.session.add(
models.UserProfile(
username="user%s" % (uuid.uuid4()),
password="password%s" % str(uuid.uuid4()),
email="%s@gamil.com" % str(uuid.uuid4())
)
)
db.session.commit() # 提交
result = db.session.query(models.UserProfile).all()
db.session.remove() # 關閉
print(result)
return "ok"