之前用flask寫了一個簡單的web數據平台，那會兒剛學flask，所以是用單腳本寫的，把所有邏輯都放到了app.py文件中。
最近看了《Flask Web開發實戰：入門、進階與原理解析》，根據里面介紹的內容，照葫蘆畫瓢，周末的時候把后端邏輯重新寫了一下 ，本文記錄下整個過程
相對最初的那一版(用flask搭建一個測試數據生成器(v1.1))，本次變更如下:
1、使用flask_restful定義視圖函數&配置路由；
2、使用Blueprint(藍圖)模塊化組織代碼結構；
3、使用工廠函數創建app實例；
4、單獨維護一些擴展，如數據庫方法 SQLAlchemy；
5、添加模型層 models，用代碼實現建表、寫入數據等操作；
工程文件目錄如上
blueprints: 藍圖目錄，用來存放自己定義的藍圖文件；
models: 模型層，存放定義庫表的操作；
static、templates: 存放靜態文件和html模版，如果使用flask自帶的jinja模版渲染語法，可以使用它，因為我前端用的vue，所以沒有用到這兩個文件；
utils: 自定義的工具類，存放抽象出來的公共操作，如pymysql的使用；
data_factory.py: 自定義的工廠函數文件
extensions.py: 擴展文件
run_main.py: 主程序入口
一、創建藍圖文件create_data.py
新增一個藍圖文件，並在其中定義視圖函數
# coding: utf-8 # author: hmk from flask import Blueprint from flask_restful import Api, Resource from flask import request import faker fake = faker.Faker(locale='zh_CN') # 初始化，指定生成中文格式數據 create_data_bp = Blueprint('create_data', __name__) # 創建一個藍本 api = Api(create_data_bp) # 使用這個藍本創建一個Api對象 class CreatePhone(Resource): # def __init__(self): # self.fake = faker.Faker(locale='zh_CN') # 初始化，指定生成中文格式數據 @staticmethod def create_phone(num): """生成電話""" phones = [fake.phone_number() for _ in range(int(num))] # 列表推導，把生成的數據組成一個列表 return " ".join(phones) def get(self): # 使用flask_restful編寫get方法時，視圖函數名必須是get num = request.args.get("num") # 獲取前端參數"num" if num == "": data = self.create_phone(5) else: data = self.create_phone(num) return_data = { "code": 200, "data": data } # time.sleep(3) return return_data class CreateId(Resource): """創建身份證id""" @staticmethod def create_id(num=5): """生成身份證""" identity_ids = [fake.ssn() for i in range(int(num))] return " ".join(identity_ids) def post(self): num = request.form.get("num") if num == "" or num is None: data = self.create_id(5) else: data = self.create_id(num) return_data = { "code": 200, "data": data } return return_data class CreateName(Resource): """創建姓名""" @staticmethod def create_name(num): """生成姓名""" names = [fake.name() for i in range(int(num))] # 生成多個 return " ".join(names) def post(self): num = request.json.get("num") if num == "": data = self.create_name(20) else: data = self.create_name(num) return_data = { "code": 200, "data": data }
return return_data # 為每個api配置路由 api.add_resource(CreatePhone, '/api/create_data/phone') api.add_resource(CreateId, '/api/create_data/id') api.add_resource(CreateName, '/api/create_data/name')
我在上述代碼中加了一些注釋
1、首先新增了一個藍圖 create_data_bp；
2、使用flask_restful定義接口時，需要先創建一個api對象，之前在創建api對象時，傳入的是app對象，但是這里我們用到了藍圖，所以需要傳入藍圖對象，為每個藍圖創建對應的視圖函數；
3、這里可以理解為創建了3個視圖函數，一個class代表一個，如上面的CreatePhone、CreateId、CreateName，這個類繼承flask_restful的 Resource類，
同時有一點要額外注意⚠️：定義get或post接口抑或其他類型的接口時，要使用對應的方法名，例如定一個get方法的接口，那么需要在這個類下新建一個方法名為get的方法（名稱不能自己隨意定義～）；
4、定義好視圖函數(接口)后，使用 api.add_resource為每個視圖添加路由；
用這種方式可以創建其他藍圖文件
二、創建工廠函數，把藍圖引入並注冊
在data_factory.py中創建一個工廠函數，並且引入我們之前建好的藍圖文件
# coding: utf-8 # author: hmk from flask import Flask from blueprints.create_data import create_data_bp from blueprints.weather import weather_bp from flask_cors import CORS def create_app(): app = Flask(__name__) CORS(app, supports_credentials=True) # 設置允許跨域 app.config.update(RESTFUL_JSON=dict(ensure_ascii=False)) # 解決flask接口中文數據編碼問題(使用RESTFUL) register_blueprints(app) return app def register_blueprints(app): """注冊藍本""" app.register_blueprint(create_data_bp) # 注冊藍本 app.register_blueprint(weather_bp)
create_app() 是我創建的工廠函數，我們需要在工廠函數中注冊之前創建好的藍圖
除了藍圖外，我們需要在工廠函數中注冊各種東西，例如初始化數據庫，所以為了便於維護，我們按照不同的作用來定義各自的函數，
例如上述代碼中定義了一個 register_blueprints 函數，傳入的參數為app對象，它的作用就是注冊藍本，
然后在工廠函數中調用這個函數即可
三、創建一個run_main.py文件，調用工廠函數創建app實例
# coding: utf-8 # author: hmk from data_factory import create_app app = create_app() if __name__ == '__main__': app.run(debug=True, host='127.0.0.1', port=5000)
這個文件中，導入了工廠函數並且生成了一個app
啟動時指定了host、端口、打開debug模式
此時，如果啟動這個文件的話(和啟動其他py文件一樣)，這個flask項目就運行起來了
除此之外還有一種方式，在終端運行命令：flask run
這種方式可以不創建run_main.py，設置 FLASK_app（指明Flask實例對象 app 所在的模塊位置）即可，FLASK_app可以放到.flaskenv文件中，也可以在執行命令前設置一下
例如我的app實例是在data_factory.py中生成的，所以需要指明 FLASK_app=data_factory（當然這里我的run_main.py中也生成了app實例，用它也行）
(1)放到.flaskenv文件中
注意：mac和windows下是不同的
mac -- FLASK_APP=data_factory
windows -- FLASK_app=data_factory
FLASK_ENV=development
FLASK_app=data_factory
然后在終端執行命令
D:\python_code_study\flaskProject>flask run
(2)如果不把它放到.flaskenv中，那么可以在終端依次執行如下命令
set FLASK_app=data_factory
flask run
四、使用SQLAlchemy，編寫model.py
1、首先創建一個存放所有擴展類的文件extensions.py，例如可以把 SQLAlchemy 的初始化工作放在這里
# coding: utf-8 # author: hmk from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() # 創建一個數據庫初始化擴展類，此時不必傳入程序實例app
2、編寫models/model.py文件
# coding: utf-8 # author: hmk from extensions import db class DataFactory(db.Model): # 繼承SQLAlchemy.Model對象，一個對象代表了一張表 id = db.Column(db.Integer, primary_key=True, autoincrement=True, unique=True) # id 整型，主鍵，自增，唯一 class_name = db.Column(db.String(50)) # 類型名稱 字符串長度為20 class_id = db.Column(db.Integer, default=20) # 類型id 整型，默認為20 start_time = db.Column(db.DateTime) # 創建時間 end_time = db.Column(db.DateTime) # 創建結束時間 __tablename__ = 'data_factory' # 該參數可選，不設置會默認的設置表名，如果設置會覆蓋默認的表名
3、在工廠函數中導入擴展類
data_factory.py
# coding: utf-8 # author: hmk from flask import Flask from blueprints.create_data import create_data_bp from blueprints.weather import weather_bp from blueprints.statistics import statistics_bp from extensions import db from flask_cors import CORS def create_app(): app = Flask(__name__) CORS(app, supports_credentials=True) # 設置允許跨域 app.config.update(RESTFUL_JSON=dict(ensure_ascii=False)) # 解決flask接口中文數據編碼問題(使用RESTFUL) app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/test' # 數據庫配置 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 是否跟蹤對象的修改，默認為None，這里設置為False register_extensions(app) register_blueprints(app) return app def register_extensions(app): """注冊擴展""" db.init_app(app) # 初始化數據庫，定義一個參數，表示程序實例app，用這個實例初始化db def register_blueprints(app): """注冊藍本""" app.register_blueprint(create_data_bp) # 注冊藍本 app.register_blueprint(weather_bp) app.register_blueprint(statistics_bp)
上述代碼中，新增了一個函數 register_extensions，在里面對db使用 init_app()方法，傳入程序實例app完成初始化操作，然后在工廠函數中引用這個函數
同時在工廠函數下進行了數據庫的一些設置
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/test' # 數據庫配置
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 是否跟蹤對象的修改，默認為None，這里設置為False做完上述操作后，還沒有把model中定義的類轉化為真正的表，打開flask shell環境，執行如下命令
D:\python_code_study\flaskProject>flask shell >>> from extensions import db >>> from models.model import * >>> db.create_all()
此時數據庫出現了我們在model.py中定義的表
五、修改create_data.py，寫入數據到數據庫中
# coding: utf-8 # author: hmk from flask import Blueprint from flask_restful import Api, Resource from flask import request import faker from datetime import datetime import time from models.model import DataFactory from extensions import db fake = faker.Faker(locale='zh_CN') # 初始化，指定生成中文格式數據 create_data_bp = Blueprint('create_data', __name__) # 創建一個藍圖 api = Api(create_data_bp) # 使用這個藍圖創建一個Api對象 class CreatePhone(Resource): @staticmethod def create_phone(num): """生成電話""" phones = [fake.phone_number() for _ in range(int(num))] # 列表推導，把生成的數據組成一個列表 return " ".join(phones) def get(self): # 使用flask_restful編寫get方法時，視圖函數名必須是get start_time = datetime.now() # 定義數據開始創建時間 num = request.args.get("num") # 獲取前端參數"num" if num == "": data = self.create_phone(5) else: data = self.create_phone(num) return_data = { "code": 200, "data": data } # time.sleep(3) end_time = datetime.now() # 定義數據創建完成時間 record = DataFactory(class_name="電話號碼", class_id="1", start_time=start_time, end_time=end_time) # 調用模型方法，寫入數據 db.session.add(record) db.session.commit() return return_data class CreateId(Resource): """創建身份證id""" @staticmethod def create_id(num=5): """生成身份證""" identity_ids = [fake.ssn() for i in range(int(num))] return " ".join(identity_ids) def post(self): start_time = datetime.now() num = request.form.get("num") if num == "" or num is None: data = self.create_id(5) else: data = self.create_id(num) return_data = { "code": 200, "data": data } end_time = datetime.now() record = DataFactory(class_name="身份證id", class_id="2", start_time=start_time, end_time=end_time) db.session.add(record) db.session.commit() return return_data class CreateName(Resource): """創建姓名""" @staticmethod def create_name(num): """生成姓名""" names = [fake.name() for i in range(int(num))] # 生成多個 return " ".join(names) def post(self): start_time = datetime.now() num = request.json.get("num") if num == "": data = self.create_name(20) else: data = self.create_name(num) return_data = { "code": 200, "data": data } end_time = datetime.now() record = DataFactory(class_name="姓名", class_id="3", start_time=start_time, end_time=end_time) db.session.add(record) db.session.commit() return return_data # 為每個api配置路由 api.add_resource(CreatePhone, '/api/create_data/phone') api.add_resource(CreateId, '/api/create_data/id') api.add_resource(CreateName, '/api/create_data/name')
上述標紅的代碼，調用創建好的model，然后把數據寫入數據庫（這里是每創建成功一條數據后，便向數據庫插入一條記錄）
重新運行一下，調一下上述定義好的3個接口，分別為：
http://127.0.0.1:5000/api/create_data/phone
http://127.0.0.1:5000/api/create_data/id
http://127.0.0.1:5000/api/create_data/phone
數據庫中也插入了對應的記錄 ：
