flask-利用Blueprint、flask_restful编写一个后端测试项目

2022-03-15 16:55:40 浏览数 (1)

之前用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

新增一个蓝图文件，并在其中定义视图函数

代码语言：javascript复制
# 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中创建一个工厂函数，并且引入我们之前建好的蓝图文件

代码语言：javascript复制
# 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实例

代码语言：javascript复制
# 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

代码语言：javascript复制
FLASK_ENV=development
FLASK_app=data_factory

然后在终端执行命令

代码语言：javascript复制
D:python_code_studyflaskProject>flask run

(2)如果不把它放到.flaskenv中，那么可以在终端依次执行如下命令

代码语言：javascript复制
set FLASK_app=data_factory
flask run

4.使用SQLAlchemy，编写model.py

1、首先创建一个存放所有扩展类的文件extensions.py，例如可以把 SQLAlchemy 的初始化工作放在这里

代码语言：javascript复制
# coding: utf-8
# author: hmk

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()  # 创建一个数据库初始化扩展类，此时不必传入程序实例app

2、编写models/model.py文件

代码语言：javascript复制
# 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

代码语言：javascript复制
# 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完成初始化操作，然后在工厂函数中引用这个函数

同时在工厂函数下进行了数据库的一些设置

代码语言：javascript复制
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql pymysql://root:123456@localhost:3306/test'  # 数据库配置
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 是否跟踪对象的修改，默认为None，这里设置为False

做完上述操作后，还没有把model中定义的类转化为真正的表，打开flask shell环境，执行如下命令

代码语言：javascript复制
D:python_code_studyflaskProject>flask shell
>>> from extensions import db
>>> from models.model import *
>>> db.create_all()

此时数据库出现了我们在model.py中定义的表

5.修改create_data.py，写入数据到数据库中

代码语言：javascript复制
# 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个接口，分别为：

代码语言：javascript复制
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

数据库中也插入了对应的记录 ：

python flask 数据库 sql sqlalchemy

0 人点赞