Flask 實現 Rest API (03)
阿新 • • 發佈：2018-12-13
Web 服務是無狀態的，那麼對於客戶端提交的請求，伺服器如何判斷請求的合法性呢？比如只能登陸成功的使用者，伺服器才按要求返回資料。本文提供一種利用 會話 (session) 機制實現儲存使用者登入資訊的方法。
會話是客戶端登入到伺服器並登出的時間間隔，對需要在此會話中儲存的資料存放在在伺服器上的臨時目錄中。伺服器端與每個客戶端的會話都有一個分配的會話ID (session ID)。
利用 session 記錄登入資訊
為了在伺服器端儲存使用者的登入資訊，需要在伺服器端儲存會話資料，瀏覽器每次請求的時候，都需要攜帶會話資料，伺服器端進行校驗。這種以使用者鑑權為目的的 session 也可也被稱為 token。大致的流程如下：
- 客戶端使用使用者名稱和密碼請求登入
- 服務端收到請求，驗證使用者名稱與密碼
- 驗證成功後，伺服器端會簽發一個 Token，再把這個 Token 傳送給客戶端
- 客戶端收到 Token 以後可以把它儲存起來，比如放在 Cookie 裡或者 Local Storage 裡
- 客戶端每次向伺服器端請求資源的時候需要帶著服務端簽發的 Token
- 伺服器端收到請求，去驗證客戶端請求裡面攜帶的 Token，如果驗證成功，就向客戶端返回請求的資料
Flask 實現 session
Flask 實現 session，對每個客戶端的會話分配一個會話 ID。 會話資料儲存在 Headers 的 cookie 中，伺服器以加密方式簽名。 對於這種加密，Flask application 需要定義一個 SECRET_KEY
app.config['SECRET_KEY'] = 'your_secret_key_here'
然後可以設定會話物件（字典型別的物件），比如在伺服器設定 username 的會話物件：
@main.route('/token', methods=['POST'])
def login():
username = request.json.get('username')
password = request.json.get('password')
remember_user = request.json. 設定 session 物件 的程式碼語句 session['username'] = username。
會話物件是一個字典物件。要刪除會話變數，使用 pop() 方法。
@main.route('/token', methods=['DELETE'])
def logout():
if 'username' in session:
session.pop('username')
return jsonify({'result': 1,'description': 'Logout successful.'})
else:
return jsonify({'result': 0, 'description': 'No user was found.'})
設定 session 過期時間
登入之後，伺服器和你的瀏覽器之間建立了一個 session，它通常有一個過期時間，比如 30 分鐘，假設登入後 10 分鐘你進行了一次操作，發出 get 請求，那麼這個過期時間就要重新計算。從你操作的這一刻起，30 分鐘以後過期。這就意味著，如果你 30 分鐘內沒有任何操作，session 就會過期，必須重新登入。
Flask 設定 session 的過期時間需要兩條語句：
app.permanent_session_lifetime = datetime.timedelta(seconds=30*60)
session.permanent = True
這樣就設定了 session 的過期時間為 30 分鐘。
完整程式碼
# token.py
# encoding: utf-8
from .. import db
def get_users():
conn = db.connect()
curr = conn.cursor()
curr.execute('select username, pwd as password from users')
users_dict = [dict((curr.description[i][0], value)
for i, value in enumerate(rows))
for rows in curr.fetchall()]
curr.close()
conn.close()
return users_dict
def get_password(user_name):
result = ''
users = get_users()
for user in users:
if user['username'].upper() == user_name.upper():
result = user.get('password')
return result
def verify_user(user_name, password):
if password == get_password(user_name):
return True
else:
return False
# token_routes.py
# encoding: utf-8
from . token import *
from . import main
from flask import request, session, jsonify
@main.route('/token', methods=['POST'])
def login():
username = request.json.get('username')
password = request.json.get('password')
remember_user = request.json.get('remember_user')
if verify_user(username, password) == True:
session['username'] = username
if remember_user == True:
session.permanent = True
else:
session.permanent = False
return jsonify({'result': 1, 'description': 'Login successful.'}), 200
else:
return jsonify({'result': 0, 'description': 'Login failed.'}), 401
@main.route('/token', methods=['DELETE'])
def logout():
if 'username' in session:
session.pop('username')
return jsonify({'result': 1,'description': 'Logout successful.'})
else:
return jsonify({'result': 0, 'description': 'No user was found.'})
路由中新增 session 校驗
在伺服器端，對相關的檢視函式 (view function）進行校驗：
@main.route('/employees')
def listEmployees():
if 'username' in session:
emp = Employee();
employees = emp.listAll()
return jsonify({'rows': employees}), 200;
else:
return jsonify({'error': 'No authorization.'}), 401
客戶端測試
使用自己喜歡的 Rest 測試工具，比如 Postman。本文提供利用 requests 庫進行請求的示例：
import requests
import json
def get_session_cookie():
result = ''
payload = {
"username": "admin",
"password": "pwd"
}
resp = requests.post('http://localhost:5000/token', json=payload)
for k, v in resp.cookies.items():
if k == 'session':
result = v
return result
cookie = get_session_cookie()
def get_employees():
session_cookie = {
'session': cookie
}
resp = requests.get('http://localhost:5000/employees', cookies=session_cookie)
return resp
def create_employee():
session_cookie = {
'session': cookie
}
payload = {
"AGE": 26,
"EDUCATION": "Lower secondary",
"EMAIL": "[email protected]",
"EMP_ID": 9001,
"GENDER": "Female",
"MARITAL_STAT": "Married",
"NR_OF_CHILDREN": 4,
"PHONE_NR": "980-0639-38"
}
resp = requests.post(
'http://localhost:5000/employees/create',
json=payload,
cookies=session_cookie
)
return resp
def modify_employee():
session_cookie = {
'session': cookie
}
payload = {
"AGE": 26,
"EDUCATION": "Lower secondary",
"EMAIL": "[email protected]",
"EMP_ID": 9001,
"GENDER": "女",
"MARITAL_STAT": "Married",
"NR_OF_CHILDREN": 4,
"PHONE_NR": "980-0639-38"
}
resp = requests.put(
'http://localhost:5000/employees/9001',
json=payload,
cookies=session_cookie)
return resp
def delete_employee():
session_cookie = {
'session': cookie
}
resp = requests.delete(
'http://localhost:5000/employees/9001',
cookies=session_cookie)
return resp