您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python Web 開發 | Flask（一）簡單實現用戶注冊、登錄、注銷、查詢接口 | 初學者案例 | 簡單封裝響應數據 | 使用 Postman 進行接口測試

編輯：Python

文章目錄

一、前置准備
- 1.1 創建數據表
- 1.2 修改 Postman 的請求頭
二、具體實現
- 2.1 utils.py
- 2.2 mysql.py
- 2.3 models.py
- 2.4 entity.py
- 2.5 app.py
三、Postman 測試接口
- 3.1 測試訪問接口
- 3.2 測試注冊接口
- - 3.2.1 參數為空
  - 3.2.2 缺少參數
  - 3.2.3 用戶名重復
  - 3.2.4 注冊成功
- 3.3 測試登錄接口
- - 3.3.1 缺少參數
  - 3.3.2 用戶名或密碼錯誤
  - 3.3.3 登錄成功
- 3.4 測試用戶注銷接口
- - 3.4.1 操作成功
  - 3.4.2 暫未登錄
- 3.4 測試用戶狀態驗證接口
- - 3.4.2 暫未登錄
  - 3.4.1 驗證通過
- 3.5 測試查詢用戶數據的接口
- - 3.5.1 測試查詢所有用戶接口
  - 3.5.2 測試查詢指定ID用戶的接口

Flask 是一個使用 Python 編寫的輕量級 Web 應用程序框架。接下來以初學者的角度來編寫後端的接口，不涉及前端代碼，即沒有用到 Flask 的模板渲染。

運行環境：

windows11
PyCharm 專業版 2021.3
MySQL 5.7.36
Navicat 15 數據庫管理工具
Postman 接口請求工具
Anaconda 3
python 3.8
flask 2.2.0
flask_sqlalchemy 2.5.1 ，flask 框架支持ORM 數據持久化的庫

案例的文件結構：

一、前置准備

1.1 創建數據表

本次小案例，只用到一張用戶表 sys_user

建表的 MySQL 語句：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for sys_user
-- ----------------------------
DROP TABLE IF EXISTS `sys_user`;
CREATE TABLE `sys_user` (
`user_id` int(11) NOT NULL AUTO_INCREMENT,
`user_name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
`user_password` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
PRIMARY KEY (`user_id`) USING BTREE,
UNIQUE INDEX `user_name`(`user_name`) USING BTREE COMMENT '用戶名稱不可重復'
) ENGINE = InnoDB AUTO_INCREMENT = 8 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
-- ----------------------------
-- Records of sys_user
-- ----------------------------
INSERT INTO `sys_user` VALUES (1, '小王', '123456');
INSERT INTO `sys_user` VALUES (2, '小綠', '123456');
INSERT INTO `sys_user` VALUES (3, 'uni', '123456');
SET FOREIGN_KEY_CHECKS = 1;

1.2 修改 Postman 的請求頭

最後在進行接口測試時，需要通過 JSON 傳值，這裡需要指定為 application/json; 的格式，為避免出現中文亂碼，最好加上charset=utf8

Content-Type = application/json;charset=utf8

二、具體實現

2.1 utils.py

Flask 框架在處理請求後，返回的數據必須是可轉化為JSON的類型，即字典類型。此工具文件中的方法支持將列表裡的所有 User 對象轉化到 dict 字典中，其中的 key 為下標，value 則為 Use對象調用 dict() 的結果，這個方法是自定義的，待會在 models.py 裡會說明。

""" 將列表裡的對象逐一轉化為字典, 對象需提供 dict 方法 """
def model_list_to_dict(modelList: list):
result = {
}
for index, model in enumerate(modelList):
result.update({
index: model.dict()})
return result

2.2 mysql.py

該腳本只是創建了 SQLAlchemy() 對象，至於為什麼將它單獨放在這個文件裡，是因為後面的 models.py 和 app.py 都將用到它，這裡類似單例模式的設計，是為了防止重復創建 SQLAlchemy() 對象

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

2.3 models.py

Flask 框架中 flask_sqlalchemy 庫支持數據庫的交互，使用前需定義數據庫中表的對應類，該類的特點有：

必須繼承 SQLAlchemy() 對象中的 Model
必須指定表名，__tablename__
定義字段對應的變量必須是靜態變量，且需調用 SQLAlchemy() 對象的 Column 方法

這裡的 dict() 方法是我自定義的，個人覺得這樣比較方便，在 Flask 返回數據時必須是能轉化成 JSON 的字符串，同時在輸出時也方便查看，相當於 Java 的 toString

from mysql import db
"""-------------------------------------- 定義 MySQL 用戶的映射類 -----------------------------------------"""
class User(db.Model):
__tablename__ = "sys_user"
id = db.Column(db.Integer, name='user_id', primary_key=True)
name = db.Column(db.String, name='user_name', unique=True)
password = db.Column(db.String, name='user_password')
def dict(self):
return {

'id': self.id,
'name': self.name,
'password': self.password
}

2.4 entity.py

前後端交互響應類 r ，在接口返回數據時調用，其實直接調用 flask 提供的 jsonify()方法就行，這裡自定義一個 r，作用是為了指定默認的參數，比如指定 code = 200，默認為 200 響應代碼，表示響應成功。除此之外，我們返回的 r 是繼承於 python內置的** dict 字典 **的，支持轉化為 JSON 格式的數據。
這種返回的 r 一定會存在 code、msg 和 data 這三個 key，方便前端對接口響應的數據進行判斷與處理。
jsonify 是備用的，如果出現報錯，再調用 flask 提供的這個方法。

from flask import jsonify
""" 前後端交互響應類 """
class r(dict):
def __init__(self, code=200, msg='空', data=None):
super().__init__(code=code, msg=msg, data=data)
def jsonify(self):
return jsonify(self)

2.5 app.py

app.py 是創建 Flask 實例以及編寫 API 接口的主要實現的文件，這裡其實應該可以分為兩個文件，由於是初學就暫時不分那麼清晰。
代碼看起來比較長，但邏輯是很清楚的，先創建 Flask 實例，使用 pymysql 加載 mysql 引擎，通過 Flask 實例對象配置 MySQL 連接的 URL（URL裡包含了數據庫的賬號和密碼），使用 @app.route() 標記負責響應請求的類，可以類比 SpringMVC 裡的 Controller。

from flask import Flask, request, session
from typing import List
from mysql import db
from models import User
from utils import model_list_to_dict # 自定義的 utils 工具包
import pymysql
from entity import r
# 創建 Flask 實例
app = Flask(__name__)
"""---------------------------------------- 初始化 Flask -------------------------------------------"""
# 初始化數據庫
pymysql.install_as_MySQLdb()
# 配置數據庫
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://root:[email protected]:3306/flask_test_db"
# 解決中文亂碼
app.config['JSON_AS_ASCII'] = False
# 關閉
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 設置 session 密鑰
app.config['SECRET_KEY'] = 'I LIKE U'
# 初始化數據庫
db.init_app(app)
"""---------------------------------------- API: 初始化 -------------------------------------------"""
@app.route('/', methods=['GET'])
def hello():
return r(code=200, msg='服務器訪問成功!')
"""---------------------------------------- API: 查詢所有用戶 -------------------------------------------"""
@app.route('/user', methods=['GET'])
def user_find_all():
userList: List[User] = User.query.all()
return r(code=200, msg='用戶查詢成功', data=model_list_to_dict(userList))
"""---------------------------------------- API: 根據ID查詢指定用戶 -------------------------------------------"""
@app.route('/user/<int:user_id>', methods=['GET'])
def user_find(user_id):
user = User.query.get(user_id)
if user == None:
return r(code=200, msg='用戶不存在')
else:
return r(code=200, msg='用戶查詢成功', data=user.dict())
"""---------------------------------------- API: 注冊用戶 -------------------------------------------"""
@app.route('/register', methods=['POST'])
def user_register():
reqJSONData = request.get_json(silent=True) # 允許 請求體的 raw 為空
# 1. 處理請求參數為空
if not reqJSONData: return r(code=401, msg='注冊失敗, 請求參數為空')
username = reqJSONData.get('username')
password = reqJSONData.get('password')
# 2. 處理請求參數缺少
if not all([username, password]):
return r(code=401, msg='注冊失敗, 缺少請求參數', )
# 3. 根據 user_name 字段判斷用戶是否注冊過
try:
user = User.query.filter_by(name=username).first()
if user:
return r(code=401, msg='注冊失敗, 用戶已存在')
except Exception as e:
print(e, '[Error] in [/user] [POST] when select MYSQL user where name=[] .')
return r(code=402, msg='服務器內部出錯')
# 4. 注冊用戶到數據庫
user = User(name=username, password=password)
try:
db.session.add(user)
db.session.commit() # 提交事務
except Exception as e:
db.session.rollback() # 異常時回滾
print(e, '[Error] in [/user] [POST] when inserting a user into MySQL.')
return r(code=402, msg='服務器內部出錯')
print('新用戶注冊成功: ', user.dict())
return r(code=200, msg='注冊成功', data=user.dict())
"""---------------------------------------- API: 用戶登錄 -------------------------------------------"""
@app.route('/login', methods=['POST'])
def user_login():
reqJSONData = request.get_json(silent=True) # 允許 請求體的 raw 為空
# 1. 處理請求參數為空
if not reqJSONData: return r(code=401, msg='注冊失敗, 請求參數為空')
username = reqJSONData.get('username')
password = reqJSONData.get('password')
# 2. 處理請求參數缺少
if not all([username, password]):
return r(code=401, msg='登錄, 缺少請求參數')
# 3. 驗證賬號和密碼
user = User.query.filter_by(name=username, password=password).first()
# 4. 用戶不存在, 直接返回
if not user:
return r(code=404, msg='用戶名或密碼錯誤')
else:
# 5. 保存用戶狀態到 session
session['user_info'] = user.dict()
return r(msg='登錄成功', data=user.dict())
"""---------------------------------------- API: 驗證用戶狀態 -------------------------------------------"""
@app.route('/login', methods=['GET'])
def user_check_login():
userDict = session.get('user_info')
if not userDict:
return r(code=401, msg='未登錄')
else:
return r(msg='用戶已登錄', data=userDict)
"""---------------------------------------- API: 注銷登錄 -------------------------------------------"""
@app.route('/logout', methods=['GET'])
def user_logout():
userDict = session.get('user_info')
if not userDict:
return r(msg='暫未登錄')
session.pop("user_info")
return r(msg='操作成功', data=userDict)
if __name__ == '__main__':
app.run(host='localhost', port=8081, debug=True)