2019-2020-1學期 自己8位學號 《網路空間安全專業導論》第四周學習總結
1.淺談FLask
“微”的含義
“微”並不代表整個應用只能塞在一個 Python 檔案內,當然塞在單一檔案內也 沒有問題。 “微”也不代表 Flask 功能不強。微框架中的“微”字表示 Flask 的目標是保持核心簡單而又可擴充套件。 Flask 不會替你做出許多決定,比如選用何 種資料庫。類似的決定,如使用何種模板引擎,是非常容易改變的。 Flask 可以 變成你任何想要的東西,一切恰到好處,由你做主。
FLask完整版<=Django #本部落格開始從web開發講解FLask,先將前後端不分離的專案,後將vue +flask,後從運維開發講解FLask從原始碼分析flask為什麼這麼牛逼 #html,網路請看我其他部落格
2.初識Flask
1.安裝Flask
pip install flask
啟動最小的hello, word
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return "<p>Hello, World!</p>"'
if __name__ == '__main__':
app.run()
- 首先我們匯入了
Flask
類。該類的例項將會成為我們的 WSGI 應用。 - 接著我們建立一個該類的例項。第一個引數是應用模組或者包的名稱。
__name__
- 然後我們使用
route()
裝飾器來告訴 Flask 觸發函式 的 URL 。 - 函式返回需要在使用者瀏覽器中顯示的資訊。預設的內容型別是 HTML ,因此字 符串中的 HTML 會被瀏覽器渲染。
2.路由
url_for()
函式用於構建指定函式的 URL。它把函式名稱作為第一個 引數。它可以接受任意個關鍵字引數,每個關鍵字引數對應 URL 中的變數。未知變數 將新增到 URL 中作為查詢引數。
from flask import Flask, url_for app = Flask(__name__) @app.route('/') def hello1(): return 'Hello World!' @app.route('/zz/<int:post_id>') def hello2(name='zzz'): return f'Hello World!{name}' @app.route('/test') def test_url_for(): # 下面是一些呼叫示例(請在命令列視窗檢視輸出的 URL): print(url_for('hello1')) # 輸出:/ print(url_for('hello2', name='213')) # /zz/123 print(url_for('test_url_for', num=2)) # 輸出:/test?num=2 return 'Test page' if __name__ == '__main__': app.run()
string |
(預設值) 接受任何不包含斜槓的文字 |
---|---|
int |
接受正整數 |
float |
接受正浮點數 |
path |
類似 string ,但可以包含斜槓 |
uuid |
接受 UUID 字串 |
3.redict重定向
from flask import Flask, redirect
app = Flask(__name__)
@app.route('/hello/')
def hello():
return redirect("/index")
#return redirect(url_for('index')) 同上面的結果是一樣的
@app.route("/index")
def index():
return "123"
if __name__ == '__main__':
app.run()
hello尾部有一個斜槓,看起來就如同一個檔案 夾。訪問一個沒有斜槓結尾的 URL 時 Flask 會自動進行重 定向,幫您在尾部加上一個斜槓
index的 URL 沒有尾部斜槓,因此其行為表現與一個檔案類似。如果訪問這 個 URL 時添加了尾部斜槓(就會得到一個 404 “未找到” 錯 誤。這樣可以保持 URL 唯一,並有助於搜尋引擎重複索引同一頁面。
4.Jinja2
舉例:
<h1>{{ username }}的個人主頁</h1>
{% if bio %}
<p>{{ bio }}</p> {# 這裡的縮排只是為了可讀性,不是必須的 #}
{% else %}
<p>自我介紹為空。</p>
{% endif %} {# 大部分 Jinja 語句都需要宣告關閉 #}
Jinja2 的語法和 Python 大致相同,你在後面會陸續接觸到一些常見的用法。在模
板裡,你需要新增特定的定界符將 Jinja2 語句和變數標記出來,下面是三種常用的
定界符:
{{ ... }} 用來標記變數。
{% ... %} 用來標記語句,比如 if 語句,for 語句等。
{# ... #} 用來寫註釋。
編寫個人主頁
模擬資料庫資料data,建立後端程式
from flask import Flask, render_template
app = Flask(__name__)
name = "zbb"
data = [
{'name': "zbb"},
{'name': "zbb"},
{'name': "zxy"},
{'name': "zxy"},
]
@app.route('/')
def hello():
return render_template('index.html', name=name, data=data)
if __name__ == '__main__':
app.run()
templates/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>{{ name }}的主頁</title>
</head>
<body>
<h2>{{ name }}的主頁</h2>
<p>{{ data|length }} Titles</p>
<ul>
{% for z in data %} {# 迭代 data 變數 #}
<li>{{ z.name }}-{{ z.age }} </li>{# 迭代 data 變數 #}
{% endfor %} {# 使用 endfor 標籤結束 for 語句 #}
</ul>
</body>
</html>
更多模板用法請看我的這篇博文
https://www.cnblogs.com/zdqc/p/11638144.html
前端知識請看
https://www.cnblogs.com/zdqc/category/1428916.html
5.static
存放靜態檔案的目錄
我們在我們的網頁中加入圖片
<link rel="icon" href="{{ url_for('static', filename='123.png') }}">
6.資料庫
Flask和Django 資料庫上的操作的區別就是ORM
Django內建ORM,ORM是什麼?
1. MVC或者MVC框架中包括一個重要的部分,就是ORM,它實現了資料模型與資料庫的解耦,即資料模型的設計不需要依賴於特定的資料庫,通過簡單的配置就可以輕鬆更換資料庫,這極大的減輕了開發人員的工作量,不需要面對因資料庫變更而導致的無效勞動
2. ORM是“**物件-關係-對映**”的簡稱。
3. 類物件--->sql--->pymysql--->mysql服務端--->磁碟,orm其實就是將類物件的語法翻譯成sql語句的一個引擎
ORM特點
符合python思想,一切接物件,只會ORM就可以操作基本上所有的關係型資料庫,如mysql, Postgres oracle等,書寫及其簡單,但是要程式碼轉換sql,效能較低,也很難寫出高效能的sql語句
想學的ORM的同學可以看看我的這篇文章
https://www.cnblogs.com/zdqc/p/11656606.html
#flask也可以使用ORM
# pip install flask-sqlalchemy
這裡使用原生sql,封裝pymysql
pip install pymysql
pmysql.py
import pymysql
class MysqlC:
def __init__(self, host, user, password, database):
self.con = None
self.arguments = {
"host": host,
"user": user,
"password": password,
"database": database,
"charset": 'utf8'
}
def post(self, sql):
with self:
try:
data = self.cursor.execute(sql)
self.con.commit()
return data
except Exception as e:
self.con.rollback()
return e
def get(self, sql, one=None):
with self:
try:
self.cursor.execute(sql)
if one:
return self.cursor.fetchone()
else:
return self.cursor.fetchall()
except Exception as e:
return e
def __enter__(self):
if self.con is None:
try:
self.con = pymysql.connect(**self.arguments)
self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
except Exception:
raise "資料庫連線失敗!"
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
self.con.close()
self.con = None
if __name__ == '__main__':
mysql = MysqlC('123', 'root', 'z21', 'white_ip')
w = mysql.get("select * from user")
print(w)
app.py
from flask import Flask, render_template
from mysql.pmysql import MysqlC
app = Flask(__name__)
pmysql = MysqlC("121.3", "root", "我好想你啊", "zzb")
name = "zbb"
@app.route('/')
def hello():
data = pmysql.get("select name,age from t")
# [{'name': 'zbb', 'age': 83}, {'name': 'zbb', 'age': 83}, {'name': 'zbb', 'age': 83}]
print(data)
return render_template('index.html', name=name, data=data)
if __name__ == '__main__':
app.run()
index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="icon" href="{{ url_for('static', filename='123.png') }}">
<title>{{ name }}的主頁</title>
</head>
<body>
<h2>{{ name }}的主頁</h2>
<p>{{ data|length }} Titles</p>
<ul>
{% for z in data %} {# 迭代 data 變數 #}
<li>{{ z.name }}-{{ z.age }} </li>{# 迭代 data 變數 #}
{% endfor %} {# 使用 endfor 標籤結束 for 語句 #}
</ul>
</body>
</html>
目錄結構
7.配置檔案
from flask import Flask,Response,render_template
app = Flask(__name__,template_folder="templates",static_folder="static")
app.config.from_object("config.settings.DevelopmentConfig")
@app.route('/login')
def login():
return "ok"
if __name__ == '__main__':
app.run()
config下的settings.py
class Config:
DEBUG = False
TESTING = False
DATABASE_URI = 'sqlite://:memory:'
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
class DevelopmentConfig(Config):
DEBUG = True
呼叫
app.config.get("引數")
預設配置
{
'DEBUG': False, 是否開啟Debug模式
'TESTING': False, 是否開啟測試模式
'PROPAGATE_EXCEPTIONS': None,
'PRESERVE_CONTEXT_ON_EXCEPTION': None,
'SECRET_KEY': None,
'PERMANENT_SESSION_LIFETIME': timedelta(days=31),
'USE_X_SENDFILE': False,
'LOGGER_NAME': None,
'LOGGER_HANDLER_POLICY': 'always',
'SERVER_NAME': None,
'APPLICATION_ROOT': None,
'SESSION_COOKIE_NAME': 'session',
'SESSION_COOKIE_DOMAIN': None,
'SESSION_COOKIE_PATH': None,
'SESSION_COOKIE_HTTPONLY': True,
'SESSION_COOKIE_SECURE': False,
'SESSION_REFRESH_EACH_REQUEST': True,
'MAX_CONTENT_LENGTH': None,
'SEND_FILE_MAX_AGE_DEFAULT': timedelta(hours=12),
'TRAP_BAD_REQUEST_ERRORS': False,
'TRAP_HTTP_EXCEPTIONS': False,
'EXPLAIN_TEMPLATE_LOADING': False,
'PREFERRED_URL_SCHEME': 'http',
'JSON_AS_ASCII': True,
'JSON_SORT_KEYS': True,
'JSONIFY_PRETTYPRINT_REGULAR': True,
'JSONIFY_MIMETYPE': 'application/json',
'TEMPLATES_AUTO_RELOAD': None,
}
8.資料庫連線池
DBUtils是Python的一個用於實現資料庫連線池的模組。
建立一批連線到連線池,供所有執行緒共享使用。
PS:由於pymysql、MySQLdb等threadsafety值為1,所以該模式連線池中的執行緒會被所有執行緒共享
如果沒有連線池,使用pymysql來連線資料庫時,單執行緒應用完全沒有問題,但如果涉及到多執行緒應用那麼就需要加鎖,一旦加鎖那麼連線勢必就會排隊等待,當請求比較多時,效能就會降低了。
加鎖正常
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
import threading
from threading import RLock
LOCK = RLock()
CONN = pymysql.connect(host='127.0.0.1',
port=3306,
user='root',
password='123',
database='pooldb',
charset='utf8')
def task(arg):
with LOCK:
cursor = CONN.cursor()
cursor.execute('select * from tb1')
result = cursor.fetchall()
cursor.close()
print(result)
for i in range(10):
t = threading.Thread(target=task, args=(i,))
t.start()
無鎖報錯
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
import threading
CONN = pymysql.connect(host='127.0.0.1',
port=3306,
user='root',
password='123',
database='pooldb',
charset='utf8')
def task(arg):
cursor = CONN.cursor()
cursor.execute('select * from tb1')
result = cursor.fetchall()
cursor.close()
print(result)
for i in range(10):
t = threading.Thread(target=task, args=(i,))
t.start()
用法
import time
import pymysql
import threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
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='123',
database='pooldb',
charset='utf8'
)
def func():
# 檢測當前正在執行連線數的是否小於最大連結數,如果不小於則:等待或報raise TooManyConnections異常
# 否則
# 則優先去初始化時建立的連結中獲取連結 SteadyDBConnection。
# 然後將SteadyDBConnection物件封裝到PooledDedicatedDBConnection中並返回。
# 如果最開始建立的連結沒有連結,則去建立一個SteadyDBConnection物件,再封裝到PooledDedicatedDBConnection中並返回。
# 一旦關閉連結後,連線就返回到連線池讓後續執行緒繼續使用。
conn = POOL.connection()
# print(th, '連結被拿走了', conn1._con)
# print(th, '池子裡目前有', pool._idle_cache, '\r\n')
cursor = conn.cursor()
cursor.execute('select * from tb1')
result = cursor.fetchall()
conn.close()
func()
例項:
import pymysql
from dbutils.pooled_db import PooledDB
class MysqlC:
def __init__(self, host, user, password, db):
self.pool = PooledDB(
creator=pymysql, # 使用連結資料庫的模組
maxconnections=200, # 連線池允許的最大連線數,0和None表示不限制連線數
mincached=10, # 初始化時,連結池中至少建立的連結,0表示不建立
blocking=True, # 連線池中如果沒有可用連線後,是否阻塞等待。True,等待;False,不等待然後報錯
ping=0,
host=host,
port=3306,
user=user,
password=password,
database=db,
charset='utf8'
)
def post(self, sql):
with self:
try:
data = self.cursor.execute(sql)
self.con.commit()
return data
except Exception as e:
self.con.rollback()
return e
def get(self, sql, one=None):
with self:
try:
self.cursor.execute(sql)
if one:
return self.cursor.fetchone()
else:
return self.cursor.fetchall()
except Exception as e:
return e
def __enter__(self):
self.con = self.pool.connection()
self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
self.con.close()
if __name__ == '__main__':
mysql = MysqlC('12193', 'root', 'zb51', 'white_ip')
w = mysql.get("select * from user")
print(w)
放在配置檔案中
app.py
from flask import Flask, Response, render_template, current_app
from mysql.mysql_con import MysqlC
app = Flask(__name__, template_folder="templates", static_folder="static")
app.config.from_object("config.settings.DevelopmentConfig")
mysql = MysqlC(**app.config.get("PYMYSQL"))
@app.route('/')
def login():
w = mysql.get("select * from user")
print(w)
return "ok"
if __name__ == '__main__':
app.run()
setting.py
import pymysql
from dbutils.pooled_db import PooledDB
class Config:
PYMYSQL = {
"host": "1193",
"user": "root",
"password": "z1",
"db": "white_ip",
}
class ProductionConfig(Config):
DATABASE_URI = 'mysql://user@localhost/foo'
class DevelopmentConfig(Config):
HOST = "123"
7.Form表單
建立一個login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<form action="" method="post" enctype="multipart/form-data">
使用者名稱:<input type="text" name="user">
密碼:<input type="password" name="pwd">
<input type="submit" value="提交">
</form>
<script></script>
</body>
</html>
resquest可以拿到請求中的資料
from flask import Flask, render_template, request
import os
app = Flask(__name__)
app.debug = True
@app.route("/login", methods=["POST", "GET"])
def login():
if request.method == "GET":
# 獲取URL中的引數,例如傳入引數:http://127.0.0.1:5000/login?id=1
print(request.args.get("id"))
# 獲取URL中的引數 轉換成 字典
print(request.args.to_dict())
# 獲取請求原始資訊
print(request.environ)
# 路由地址 /login
print(request.path)
# 獲取訪問路徑
print(request.url) # http://127.0.0.1:5000/login?id=1
# 獲取URL頭,不包含引數 /login
print(request.base_url) # http://127.0.0.1:5000/login
return render_template("login.html")
if request.method == "POST":
# 請求頭中的資料
print(request.headers)
print(request.json) # 請求頭中 Content-type:application/json 資料序列化 request.json
print(request.data) # 請求頭中 Content-type 不包含 Form or data
# Formdata 和 Args查詢引數 中的資料
print(request.values)
username = request.form.get("user")
password = request.form.get("pwd")
print(username, password)
return "200 OK"
if __name__ == '__main__':
app.run()
8.檔案上傳
用 Flask 處理檔案上傳很容易,只要確保不要忘記在您的 HTML 表單中設定 enctype="multipart/form-data"
屬性就可以了。否則瀏覽器將不會傳送您的檔案。
from flask import request
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
f = request.files['the_file']
f.save('/var/www/uploads/uploaded_file.txt')
...
如果想要知道檔案上傳之前其在客戶端系統中的名稱,可以使用 filename
屬性。但是請牢記這個值是 可以偽造的,永遠不要信任這個值。如果想要把客戶端的檔名作為伺服器上的檔名, 可以通過 Werkzeug 提供的 secure_filename()
函式:
from werkzeug.utils import secure_filename
@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
if request.method == 'POST':
file = request.files['the_file']
file.save(f"/var/www/uploads/{secure_filename(file.filename)}")
...
9.Session
flask會將你的SessionID存放在客戶端的Cookie中
基本用法,具體用法請看練習題
使用在app.py中引入
from flask import session
.......
app = Flask(__name__)
app.secret_key = "i am session" #這裡只需要隨機指定一個字串即可
.......
session["username"] = USER["username"] #設定session
驗證session
session.get("username"): #獲取session
session.pop('username') #刪除session
11.練習
將今天所學全部用上,寫一個使用者註冊,登入驗證模組
加上密碼驗證
建立login.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="icon" href="{{ url_for('static', filename='123.png') }}">
</head>
<body>
{{ mes }}
<form action="" method="post" enctype="multipart/form-data">
使用者名稱:<input type="text" name="username">
密碼:<input type="password" name="pwd">
<input type="submit" value="提交">
</form>
</body>
</html>
reg.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<link rel="icon" href="{{ url_for('static', filename='123.png') }}">
</head>
<body>
{{ mes }}
<form action="" method="post" enctype="multipart/form-data">
使用者名稱:<input type="text" name="username">
密碼:<input type="password" name="pwd">
<input type="submit" value="註冊">
</form>
</body>
</html>
app.py
from flask import Flask, redirect, url_for, session, request, render_template
from mysql.pmysql import MysqlC
from werkzeug.security import generate_password_hash, check_password_hash #hash加密密碼和解密密碼
app = Flask(__name__)
app.secret_key = "qwerty" # 設定金鑰使用session
app.config.from_object("conf.settings.DevelopmentConfig") # 獲取配置檔案中的mysql連結
mysql_con = app.config.get("MYSQL")
pmysql = MysqlC(**mysql_con) # 調取封裝好的pymysq
@app.route('/', methods=["get", "post"])
def login():
# 獲取使用者提交的資料
if request.method == "GET":
return render_template("login.html")
if request.method == "POST":
username = request.form.get("username")
pwd = request.form.get("pwd")
if not username or not pwd:
return render_template("login.html", mes="請輸入使用者名稱密碼")
password = pmysql.get(f"select pwd from user where username='{username}'", one=True)
if password:
if check_password_hash(password["pwd"], pwd):
# 將使用者名稱加到session中用於判斷使用者是否登入了
session["username"] = username
return redirect('/index')
else:
return render_template("login.html", mes="密碼輸入錯誤")
else:
return render_template("login.html", mes="使用者名稱輸入錯誤")
@app.route('/reg', methods=["get", "post"])
def reg():
if request.method == "GET":
return render_template("reg.html")
if request.method == "POST":
username = request.form.get("username")
pwd = request.form.get("pwd")
if not username or not pwd:
return render_template("reg.html", mes="請輸入使用者名稱密碼")
#進行密碼加密
password = generate_password_hash(pwd)
try:
pmysql.post(f"insert user(username,pwd) values('{username}','{password}')")
return redirect(url_for('login'))
except Exception as e:
return render_template("reg.html", mes="註冊失敗")
@app.route("/index")
def index():
# 如果使用者登入成功
if session.get("username"):
return "使用者登陸成功!"
# 未登入跳轉到登陸頁面
else:
# 未登入跳轉到登陸頁面
return redirect('/')
if __name__ == '__main__':
app.run()
封裝好的pmysql.sql
import pymysql
class MysqlC:
def __init__(self, host, user, password, database):
self.con = None
self.arguments = {
"host": host,
"user": user,
"password": password,
"database": database,
"charset": 'utf8'
}
def post(self, sql):
with self:
self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
try:
data = self.cursor.execute(sql)
self.con.commit()
return data
except Exception as e:
self.con.rollback()
return e
def get(self, sql, one=None):
with self:
self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
try:
self.cursor.execute(sql)
if one:
return self.cursor.fetchone()
else:
return self.cursor.fetchall()
except Exception as e:
return e
def __enter__(self):
if self.con is None:
try:
self.con = pymysql.connect(**self.arguments)
except Exception:
return "資料庫連線失敗!"
return self.con
def __exit__(self, exc_type, exc_val, exc_tb):
self.cursor.close()
self.con.close()
self.con = None
settings.py
class DevelopmentConfig:
MYSQL = {
"host": "121.423",
"user": "root",
"password": "zbb521",
"database": "zzb"
}
目錄結構
熱愛技術,享受生活,感謝推薦!