HTTP通訊過程
浏覽器 -> HTTP請求
-> TCP傳輸 -> 伺服器
-> 接收請求 -> 解析封包
-> 路由分發 -> 執行視圖函數
-> 響應封包 -> HTTP響應
-> 浏覽器
前端和後端
前端:
-浏覽器
-手機app
-爬蟲
-程式urllib urllib2 ajax
後端:
伺服器 + WSGI + web架構程式(路由+視圖)
uWSGI Django(進行路由分發,定義視圖函數)
process=4
thread=2
伺服器模型(并發)
-多程序
-多線程
-協程
web架構程式
重量級架構:Django
輕量級架構:
Flask, Tornado
Django
提供了一站式解決方案
內建了 MVT(model view template)
ORM
路由:Werkzeug工具箱
模闆:Jinja2
Flask擴充包
Flask-SQLalchemy 操作資料庫
Flask-migrate 遷移資料庫
Flask-Mail 郵件
Flask-WTF 表單
Flask-script 腳本
Flask-Login 認證使用者狀态
Flask-RESTful rest api
Flask-Bootstrap 前端架構Bootstrap
Flask-Moment 本地化日期和時間
虛拟環境
如果使用sudo就不是目前虛拟環境了
-p path 路徑
pip freeze > requirement.txt
pip install -r requirement.txt
Flask配置啟動
不要使用
from flask import *
__name__
目前子產品名字
flask預設傳入檔案名所在目錄為根目錄,找不到子產品則目前檔案所在目錄為根目錄
1、Flask初始化參數
import_name 導入路徑
static_url_path 靜态資源url字首
static_folder 靜态資源目錄 預設static
template_folder 模闆檔案目錄 預設templates
2、Falsk配置參數設定
# (1)使用配置檔案
DEBUG=True > config.cfg
app.config.from_pyfile("config.cfg")
# (2)使用對象配置參數
class Config(Object):
DEBUG=True
app.config.from_object(Config)
# (3)直接操作config字典對象
app.config["DEBUG"] = True
3、Falsk配置參數讀取
# (1)直接操作config字典對象
app.config.get("DEBUG")
# (2)通過current_app擷取
current_app.config.get("DEBUG")
4、Falsk啟動程式
區域網路和外網都可以通路
app.run(host="0.0.0.0", port=5000, debug=True)
路由視圖
1、檢視flask路由資訊
app.url_map
2、設定請求方式
@app.route("/", methods=["GET", "POST"])
路由相同,路徑和請求方式相同,先定義先通路
路由不同,路徑和請求方式不同
3、一個視圖多個路由
不同路由可以通過裝飾器疊加,添加到一個視圖函數之上
@app.route("/1")
@app.route("/2")
def view():
pass
4、重定向
from flask import redirect, url_for
# 視圖反推路徑
url = url_for("index")
# 重定向
redirect("/")
5、路由參數提取
@app("/<int:uid>")
def view(uid):
pass
# 接受參數:預設string(除了斜線/), int, float, path
6、自定義萬能路由轉換器
# (1)繼承轉換器類
from werkzeug.routing import BaseConverter
class RegexConverter(BaseConverter):
def __init__(self, url_map, regex):
# 調用父類的初始化方法
super(RegexConverter, self).__init__(url_map)
# 将正規表達式的參數儲存到對象屬性中
# flask會去使用這個屬性進行路由的正則比對
self.regex = regex
def to_python(self, value):
return value
def to_url(self, value):
return value
# (2)将自定義的轉換器添加到flask應用中
app.url_map.converters["re"] = RegexConverter
# (3)使用自定義轉換器路由比對 to_python
#/13512345678
@app.route("/<re(r'1[34578]\d{9}'):mobile>")
def send(mobile):
pass
# (4)視圖轉換為路由url to_url
@app.route("/")
def index():
url = url_for("send", mobile="13511111111")
return redirect(url)
request對象
from flask import request
屬性
data 請求體資料(除表單外) string
form 表單資料 MultiDict
args url查詢參數 MultiDict
cookies cookies資訊 Dict
headers 封包頭 EnvironHeaders
method HTTP方法 GET/POST
url url位址 string
files 上傳的檔案 *
前端參數
(1)form表單格式 POST
application/x-www-form-urlencoded
name=“Tom”&age=23
request.form.get() 同鍵多值取第一個
request.form.getlist() 取出同鍵所有值
(2)多媒體格式 POST
enctype=“multipart/form-data”
file = request.files.get(“file”)
file.save(“file.txt”)
(3)raw字元串
(4)binary
(5)args查詢字元串 QueryString GET
?key=value&key=value
request.args.get()
python2字元串類型
a=“中國” str “utf-8” “gbk” “ascii”
a=u"中國" unicode
“中國 %s” % “abc”
ASCII cannot decode \xxx\xxx
->
u"中國 %s" % “abc”
上下文管理器
with open("test.txt", "w") as f:
f.write("something")
自定義上下文管理器
class Foo(object):
def __enter__(self):
pass
def __exit__(self, exc_type, exc_value, exc_traceback):
pass
with Foo() as f:
pass
錯誤處理
1、abort方法
可以終止視圖函數執行,并傳回給前端特定資訊
from flask import abort
# (1)傳遞标準狀态碼資訊
abort(400)
# (2)傳遞響應體資訊
from flask import Response
abort(Response("message"))
2、自定義錯誤處理
@app.errorhandler(404)
def handle_404_error(err):
return ""
響應資訊
1、元組
響應體, 狀态碼(可以是非标), 響應頭
return "message", 404, [("language": "python")]
return "message", 404, {"language": "python"}
return "message", "404 status message", {"language": "python"}
2、make_response
from flask import make_response
resp = make_response("message") # 響應體
resp.status = "666 status" # 狀态碼
resp.headers["language"] = "python" # 響應頭
return resp
3、傳回json
json就是字元串
原理:json.dumps() + Content-Type: application/json
#(1)方式一
return jsonify({"name": "Tom"})
#(2)方式二
return jsonify(name=Tom")
cookie
1、設定cookie
預設臨時cookie,浏覽器關閉就失效,max_age設定有效期,機關:秒s
from flask import make_response
resp = make_response("message")
#(1)通過原始方式設定
resp.set_headers["Set-Cookie"] = "key=value;"
#(2)通過flask提供的方式設定
resp.set_cookie("key", "value", max_age=3600)
return resp
2、擷取cookie
from flask import request
value = request.cookies.get("key")
3、删除cookie
設定cookie的過期時間為目前時間,讓浏覽器自己删除
from flask import make_response
resp = make_response("message")
resp.delete_cookie("key", "value")
return resp
session
廣義:機制
狹義:儲存到伺服器中的session資料
浏覽器 <- cookie(session_id) -> 伺服器 <—> 資料庫MySQL
session儲存的地方:
MySQL、Redis、MongoDB、檔案、程式記憶體(不支援多伺服器)
flask預設把session儲存到了cookie中, 不安全
隻使用session_id
(1)儲存到cookie
(2)儲存到url中,具有局限性,适用于單次通路
# 1、設定session秘鑰, 加密混淆,防止session被篡改
app.config["SECRET_KEY"] ="ooxx"
# 2、設定session資料
from flask import session
session["key"] = "value"
# 3、擷取session資料
value = session.get("key")
請求上下文與應用上下文
(1)請求上下文
request, session對象都是全局對象
全局變量 - 線程局部變量
(2)應用上下文
current_app 目前運作程式檔案的程式執行個體
g 處理請求時,用于臨時存儲的對象,每次請求都會重置,用于一次請求多個函數傳參
g.name = "張三"
請求鈎子hook
類似Django中的中間件
1、處理第一個請求前執行
@app.before_first_request
def handle():
pass
2、處理每個請求前執行
@app.before_request
def handle():
pass
3、處理每個請求後執行,視圖函數沒有出現異常執行
@app.after_request
def handle(response):
return response
4、處理每個請求後執行,視圖函數是否出現異常都執行, 非調試模式debug=False
@app.teardown_request
def handle(response):
return response
使用
app.pyrunserver -h 0.0.0.0 -p 8000
python
shell
jinja2模闆
1、渲染
from flask import render_template
data = {
"name": "python",
"age": 18,
"my_dict": {"city": "beijing"},
"my_list": [1, 2, 3, 4, 5],
}
render_template("index.html", **data)
變量 {undefined{ name }}
字典 {undefined{ my_dict.city }} {undefined{ my_dict[“city”] }}
清單 {undefined{ my_list[0] }}
數字相加 {undefined{ 1 + 2 }}
字元串拼接 {undefined{ “hello” + “world” }}
2、 字元串過濾器(支援鍊式調用 )
safe 禁用轉義 xss攻擊
capitalize 首字母大寫,其他小寫
lower 轉小寫
upper 轉大寫
title 每個單詞首字母大寫
trim 首尾空格去掉
reverse 反轉字元串
format 格式化輸出(類似%格式化)
striptags 删除html标簽
3、清單過濾器
first 取第一個元素
last 取最後一個元素
length 擷取清單長度
sum 清單求和
sort 清單排序
4、自定義過濾器
方式一:
(1)定義過濾器
def list_step_2(lst):
return lst[::2]
(2)注冊過濾器
app.add_template_filter(list_step_2, "list_step_2")
(3)使用過濾器
{{ lst | list_step_2 }}
方式二:
@app.template_filter("list_step_2")
def list_step_2(lst):
return lst[::2]
模闆宏macro
1、不帶參數宏
{% macro input() %}
<input type="text" >
{% endmacro %}
{{ input() }}
2、帶參數宏
{% macro input(type) %}
<input type="{{type}}" >
{% endmacro %}
3、帶預設參數宏
{% macro input(type="text") %}
<input type="{{type}}" >
{% endmacro %}
4、導入宏
{% import "macro.html" as my_macro%}
{{ my_macro.input() }}
模闆繼承
bash.html
{% block top%}
父模闆的内容
{% endblock %}
child.html
{% extends "bash.html" %}
{% block top%}
子模闆的内容
{% endblock %}
模闆包含
{% inclued "hello.html" %}
中的特殊變量
可以直接在模闆中使用
config
request
url_for
get_flashed_messages基于session
from flask import flash
flash("hello1")
flash("hello2")
{% for msg in get_flashed_messages() %}
{{msg}}
{% endfor %}
Flask-Script擴充
pip install Flask-Script
Manager 啟動指令管理類
from flask import Flask
from flask_script import Manager
app = Flask(__name__)
#建立app管理類
manager = Manager(app)
@app.route("/")
def index():
pass
if __name__ == "__main__":
# app.run()
manager.run()
表單擴充
pip install Flask-WTF
可以進行csrf驗證 表單生成 和 驗證
Flask資料庫擴充
pip install Flask-SQLAlchemy
Flask資料庫遷移擴充
pip install Flask-Script (依賴)
pip install Flask-Migrate
Flask發送郵件
pip install Flask-Mail
循環導入-死鎖
解決:一方讓步,延遲導入
路由設定的方式
1、裝飾器
@app.route("/")
def view():
pass
2、裝飾器原始方式
def view():
pass
app.route("/")(view)
藍圖Blueprint
一個小子產品的抽象概念, 延遲加載
1、定義藍圖
# __init__.py
from flask import Blueprint
# 工程目錄的優先級大于應用目錄
app_user = Blueprint("app_user", __name__, template_folder="templates")
# init執行的時候把視圖加載進來
from .views import view
2、定義視圖
# views.py
from . import app_user
@app_user.route("/")
def view():
pass
3、注冊藍圖
# main.py
from flask import Flask
app = Flask(__name__)
app.register_blueprint(app_user, url_prefix="/user")
4、檢視路由
app.url_map
單元測試
web開發階段:
需求分析,設計階段,實作階段,測試階段
測試階段
單元測試,內建測試,系統測試
斷言assert
-真 繼續執行
-假 抛出異常 AssertionError
Flask單元測試
app.test_client()
測試資料庫
開啟->測試->斷開
Flask部署
pip install gunicorn
gunicorn
-w 4 // 程序數
-b 127.0.0.1:5000 // 開啟端口
--access-logfile ./logs //日志檔案
-D // 守護程序方式運作
main:app // 要運作的程式入口
ps aux|grep gunicorn
Nginx
負載均衡
靜态檔案
cp nginx.conf nginx.conf.bak
upstream flask{
server 10.0.0.1:5000;
server 10.0.0.1:5001;
}
server{
listion 80;
server_name localhost;
location / {
proxy_pass http://flask;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr
}
平滑重新開機 nginx -s reload
mysql異常
@@tx_isolation
修改報錯檔案base.py# cursor.execute("SELECT @@tx_isolation")
if self.server_version_info < (5, 7, 20):
cursor.execute("SELECT @@tx_isolation")
else:
cursor.execute("SELECT @@transaction_isolation")