flask-利用Blueprint、flask_restful编写一个后端测试项目
flask-利用Blueprint、flask_restful编写一个后端测试项目
之前用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: 主程序入口
1. 创建蓝图文件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为每个视图添加路由；
用这种方式可以创建其他蓝图文件
2. 创建工厂函数，把蓝图引入并注册
在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对象，它的作用就是注册蓝本，
然后在工厂函数中调用这个函数即可
3. 创建一个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 run4.使用SQLAlchemy，编写model.py
1、首先创建一个存放所有扩展类的文件extensions.py，例如可以把 SQLAlchemy 的初始化工作放在这里
# coding: utf-8
# author: hmk
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy() # 创建一个数据库初始化扩展类，此时不必传入程序实例app2、编写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中定义的表
5.修改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
数据库中也插入了对应的记录 ：
