Flask基础知识

本章节先介绍flask的基础知识，以最快的方式上手flask，这里不讲那么多理论，直接实战，大家按照步骤操作即可。

注意：如无特殊说明，所有命令默认都是在Git Bash命令行工具下执行，且默认都是在激活的虚拟环境下执行。

创建虚拟环境

在使用Python语言的时候我们使用pip来安装第三方包，但是由于pip的特性，系统中只能安装每个包的一个版本。但是在实际项目开发中，不同项目可能需要第三方包的不同版本，Python的解决方案就是虚拟环境。顾名思义，虚拟环境就是虚拟出来的一个隔离的Python环境，每个项目都可以有自己的虚拟环境，用pip安装各自的第三方包，不同项目之间也不会存在冲突。

打开Git Bash

Git Bash

mkdir flaskapp

cd flaskapp

创建虚拟环境

python -m venv ./venv

激活虚拟环境

source venv/Scripts/activate

退出虚拟环境（按需）

deactivate

安装Flask依赖库

进入工程目录

安装虚拟环境

source venv/Scripts/activate

安装Flask依赖库

# 安装最新版本
pip3 install Flask
# 或指定安装版本
pip3 install Flask==2.0.2

安装依赖截图

创建Flask启动程序

创建一个文件，命名为app.py

from flask import Flask

app = Flask(__name__)


@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

设置环境变量

# 对应app.py文件名
export FLASK_APP=app
# 开发者模式
export FLASK_ENV=development
# 调试模式
export FLASK_DEBUG=1

启动Flask服务

flask run
# 或者
python app.py

命令截图

访问Flask服务

打开浏览器，输入地址：http://127.0.0.1:5000/或者http://localhost:5000/

访问截图

停止Flask服务

可以通过快捷键Ctrl+C来停止

Flask路由规则

Flask路由规则是Flask程序中用于定义URL与视图函数之间的映射关系，通过路由规则，可以定义URL与视图函数之间的映射关系，从而实现URL与视图函数的关联。

修改app.py文件如下：

from flask import Flask

app = Flask(__name__)


@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"


@app.route("/user/info")
def user_info():
    return "<p>user_info-GET请求<p>"


@app.route("/user/list", methods=['GET', 'POST'])
def user_list():
    return "<p>user_list-GET/POST请求<p>"


@app.route("/user/save", methods=['POST'])
def user_save():
    return "<p>user_save-POST请求<p>"


@app.route("/user/update", methods=['POST'])
def user_update():
    return "<p>user_update-POST请求<p>"


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问服务

# 访问/user/info
curl http://localhost:5000/user/info
# 访问/user/list
curl http://localhost:5000/user/list
# 或者
curl -X POST http://localhost:5000/user/list
# 访问/user/save
curl -X POST http://localhost:5000/user/save
# 访问/user/update
curl -X POST http://localhost:5000/user/update
# 访问一个不存在的路由
curl http://localhost:5000/user/delete
# 访问一个存在的路由但不存在的请求方式
curl http://localhost:5000/user/update -X GET

效果图 4. 使用Postman接口测试工具访问服务

访问/user/info
GET 访问/user/list
POST 访问/user/list
访问/user/save
访问/user/update
访问一个不存在的路由
访问一个存在的路由但不存在的请求方式

Flask参数传递

Flask参数传递是Flask程序中用于传递参数的机制，通过参数传递，可以传递参数给视图函数，从而实现参数的传递。

修改app.py文件如下：

from flask import Flask, request

app = Flask(__name__)


@app.route("/")
def hello_world():
    return "<p>Hello, World!</p>"


@app.route("/user/info")
def user_info():
    # url参数/user/info?userId=1
    userId = request.args.get('userId')
    return f"userId:{userId}"


@app.route("/user/list", methods=['GET', 'POST'])
def user_list():
    # /user/list?pageNum=1&pageSize=10
    pageNum = request.args.get("pageNum")
    pageSize = request.args.get("pageSize")
    # 表单参数
    userName = request.form.get("userName")
    userType = request.form.get("userType")
    return f"pageNum:{pageNum},pageSize:{pageSize},userName:{userName},userType:{userType}"


@app.route("/user/save", methods=['POST'])
def user_save():
    # 请求正文参数1
    data = request.get_data()
    return f"data：{data}"


@app.route("/user/update", methods=['POST'])
def user_update():
    # 请求正文参2
    data = request.get_data()
    return f"data:{data}"


@app.route("/user/get/<userId>", methods=['GET'])
def user_get(userId):
    # 路由参数
    return f"userId:{userId}"


@app.route("/user/get/<int:userId>", methods=['GET'])
def user_get_int(userId):
    # 路由参数-int类型
    return f"int-userId:{userId}"


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问服务

url参数

curl http://localhost:5000/user/info?userId=1

表单参数

curl -d "userName=admin&userType=1" -X POST \
http://localhost:5000/user/list?pageNum=1\&pageSize=10

请求正文参数1

curl -d "userName=admin&userType=1" -X POST \
http://localhost:5000/user/save

请求正文参数2

curl -H 'Content-Type: application/json' \
-d '{"userName":"admin","userType":1}' -X POST \
http://localhost:5000/user/update

路由参数

curl http://localhost:5000/user/get/abc

路由参数-int类型

curl http://localhost:5000/user/get/1

使用Postman接口测试工具访问接口（略）

Flask蓝图

当我们的接口服务太多，只放在一个文件上肯定是不方便的，所以需要有个模块化的方案，Flask里面使用的蓝图来实现。蓝图是Flask程序中用于模块化的机制，通过蓝图，可以模块化Flask程序，从而实现模块化的功能。

创建控制层模块

mkdir controllers

新增controllers/__init__.py文件内容无
新增controllers/user_controller.py文件

from flask import Blueprint

user = Blueprint('user', __name__, url_prefix="/sys/user")


@user.route("/detail", methods=['POST'])
def user_detail():
    return "user_detail"


@user.route("/page", methods=['POST'])
def user_page():
    return "user_page"


@user.route("/save", methods=['POST'])
def user_save():
    return "user_save"


@user.route("/update", methods=['POST'])
def user_update():
    return "user_update"


@user.route("/remove", methods=['POST'])
def user_remove():
    return "user_remove"

新增controllers/role_controller.py文件

from flask import Blueprint

role = Blueprint('role', __name__, url_prefix="/sys/role")


@role.route("/detail", methods=['POST'])
def role_detail():
    return "role_detail"


@role.route("/page", methods=['POST'])
def role_page():
    return "role_page"


@role.route("/save", methods=['POST'])
def role_save():
    return "role_save"


@role.route("/update", methods=['POST'])
def role_update():
    return "role_update"


@role.route("/remove", methods=['POST'])
def role_remove():
    return "role_remove"

修改app.py文件如下：

from flask import Flask

from controllers.role_controller import role
from controllers.user_controller import user

app = Flask(__name__)

# 注册用户模块
app.register_blueprint(user)
# 注册角色模块
app.register_blueprint(role)

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问服务

用户模块

# 添加用户
curl -d "userName=admin&userType=1" -X POST \
http://localhost:5000/sys/user/save
# 修改用户
curl -d "userName=admin&userType=1" -X POST \
http://localhost:5000/sys/user/update
# 删除用户
curl http://localhost:5000/sys/user/remove
# 查询用户详情
curl http://localhost:5000/sys/user/detail
# 分页查询用户
curl http://localhost:5000/sys/user/page?pageNum=1\&pageSize=10

角色模块

# 添加角色
curl -d "roleName=admin&roleType=1" -X POST \
http://localhost:5000/sys/role/save
# 修改角色
curl -d "roleName=admin&roleType=1" -X POST \
http://localhost:5000/sys/role/update
# 删除角色
curl http://localhost:5000/sys/role/remove
# 查询角色详情
curl http://localhost:5000/sys/role/detail
# 分页查询角色
curl http://localhost:5000/sys/role/page?pageNum=1\&pageSize=10

使用ApiFox/Postman接口测试工具访问（略）

Flask全局请求拦截

Flask的全局请求拦截器，可以拦截所有的请求，并对请求进行预处理和后处理。在Flask中，可以使用before_request和after_request装饰器来实现全局请求拦截。

修改app.py文件如下：

from flask import Flask, request

from controllers.role_controller import role
from controllers.user_controller import user

app = Flask(__name__)

# 注册用户模块
app.register_blueprint(user)
# 注册角色模块
app.register_blueprint(role)


@app.before_request
def auth():
    WHITE_LIST = ['/sys/user/detail', '/sys/user/page']
    # 简单处理一下，非白名单路由，提示登录
    if request.path not in WHITE_LIST:
        return "请先登录"


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问（略）
使用ApiFox/Postman接口测试工具访问（略）

Flask统一异常处理

Flask统一异常处理，可以统一处理所有的异常，并对异常进行预处理和后处理。在Flask中，可以使用@app.errorhandler装饰器来实现统一异常处理。

修改app.py文件如下：

from flask import Flask, request

from controllers.role_controller import role
from controllers.user_controller import user

app = Flask(__name__)

# 注册用户模块
app.register_blueprint(user)
# 注册角色模块
app.register_blueprint(role)


@app.before_request
def auth():
    WHITE_LIST = ['/sys/user/detail', '/sys/user/page', '/ex/test', '/ex/other']
    # 简单处理一下，非白名单路由，提示登录
    if request.path not in WHITE_LIST:
        return "请先登录"


@app.route("/ex/test")
def ex_test():
    """
    测试其他异常
    :return:
    """
    a = 3 / 0
    return a


@app.errorhandler(404)
def error_404(e):
    """
    404异常处理
    :param e:
    :return:
    """
    return "请求地址不存在"


@app.errorhandler(Exception)
def error(e):
    """
    其他异常处理
    :param e: 异常
    :return:
    """
    return str(e)


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问（略）
使用ApiFox/Postman接口测试工具访问（略）

Flask连接Mysql数据库

做后端服务，操作数据库是必不可少的环节，这里简单介绍一下Flask是如何连接Mysql数据库的。本文默认Mysql已经安装好的前提下进行的。

新建数据库

新建数据库

新建表

CREATE TABLE `sys_user` (
  `id` bigint(20) NOT NULL COMMENT '用户ID',
  `user_name` varchar(32) NOT NULL COMMENT '用户名',
  `real_name` varchar(32) NOT NULL COMMENT '姓名',
  `nick_name` varchar(32) DEFAULT NULL COMMENT '昵称',
  `avatar` varchar(100) DEFAULT NULL COMMENT '用户头像',
  `password` varchar(64) DEFAULT NULL COMMENT '用户密码',
  `salt` varchar(10) DEFAULT NULL COMMENT '密码加盐',
  `mobile_phone` varchar(11) NOT NULL COMMENT '手机号',
  `tel` varchar(20) DEFAULT NULL COMMENT '联系电话',
  `email` varchar(32) DEFAULT NULL COMMENT '邮箱',
  `admin_type` int(11) DEFAULT NULL COMMENT '管理员类型<sys_admin_type>',
  `sex` int(11) DEFAULT NULL COMMENT '性别<sys_sex>',
  `is_locked` tinyint(1) DEFAULT '0' COMMENT '是否锁定',
  `dept_id` bigint(20) DEFAULT NULL COMMENT '所属部门',
  `post_id` bigint(20) DEFAULT NULL COMMENT '所属岗位',
  `remark` varchar(255) DEFAULT NULL COMMENT '备注',
  `create_time` datetime(3) DEFAULT NULL COMMENT '创建时间',
  `create_user` bigint(20) DEFAULT NULL COMMENT '创建用户',
  `update_time` datetime(3) DEFAULT NULL COMMENT '更新时间',
  `update_user` bigint(20) DEFAULT NULL COMMENT '更新用户',
  `is_deleted` tinyint(1) DEFAULT '0' COMMENT '是否删除',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `idx_sys_user_user_name` (`user_name`) USING BTREE,
  KEY `idx_sys_user_real_name` (`real_name`) USING BTREE,
  KEY `idx_sys_user_mobile_phone` (`mobile_phone`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户';

插入一条记录

INSERT INTO `flaskapp`.`sys_user`(`id`, `user_name`, `real_name`, `nick_name`, `avatar`, `password`, `salt`, `mobile_phone`, `tel`, `email`, `admin_type`, `sex`, `is_locked`, `dept_id`, `post_id`, `remark`, `create_time`, `create_user`, `update_time`, `update_user`, `is_deleted`) VALUES (1567762634947375105, 'admin', '普通管理员', NULL, NULL, 'e82098b6d649e98d3c4bc7731baab8c8', 'az3xjbk0', '18600000001', '19833173224', 't.eeq@kcshc.nc', 2, 1, 0, 1575377096640569346, 1575379311115640834, NULL, '2022-09-08 14:31:54.450', 0, '2023-02-10 14:22:38.455', 1567738052492341249, 0);

激活虚拟环境

source venv/bin/activate

安装SQLAlchemy

# 安装最新版本
pip3 install SQLAlchemy
# 指定安装版本
pip3 install SQLAlchemy==1.4.26

安装SQLAlchemy 6. 安装Flask-SQLAlchemy

# 安装最新版本
pip3 install Flask-SQLAlchemy
# 指定安装版本
pip3 install Flask-SQLAlchemy==2.5.1

安装Flask-SQLAlchemy 7. 安装pymysql

# 安装最新版本
pip3 install pymysql
# 指定安装版本
pip3 install pymysql==1.0.2

安装pymysql 8. 创建模型目录

mkdir models

新增models/__init__.py文件


from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

修改app.py文件如下：

from urllib import parse

from flask import Flask, request

from controllers.role_controller import role
from controllers.user_controller import user
from modeles import db

DB_HOST = "localhost"
DB_PORT = "3306"
DB_NAME = "flaskapp"
DB_USER = "root"
DB_PASSWORD = "root"
# 防止密码中有特殊字符，需要使用parse.quote_plus进行转义
SQLALCHEMY_DATABASE_URI = f'mysql+pymysql://{DB_USER}:{parse.quote_plus(DB_PASSWORD)}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4'

app = Flask(__name__)

# 注册用户模块
app.register_blueprint(user)
# 注册角色模块
app.register_blueprint(role)
# Flask中新增数据库配置
app.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
app.config['SQLALCHEMY_RECORD_QUERIES'] = True
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
        'pool_size': 5,
        'pool_timeout': 90,
        'pool_recycle': 7200,
        'max_overflow': 1024
    }

# 初始化db
db.init_app(app)


@app.before_request
def auth():
    WHITE_LIST = ['/sys/user/detail', '/sys/user/page', '/ex/test', '/db/test']
    # 简单处理一下，非白名单路由，提示登录
    if request.path not in WHITE_LIST:
        return "请先登录"


@app.route("/db/test")
def db_test():
    cursor = db.session.execute('select * from t_user')
    result = cursor.fetchall()
    if len(result) > 0:
        u = result[0]
        return f"id:{u.id},user_name:{u.user_name}"
    return "无记录"


@app.route("/ex/test")
def ex_test():
    """
    测试其他异常
    :return:
    """
    a = 3 / 0
    return a


@app.errorhandler(404)
def error(e):
    """
    404异常处理
    :param e:
    :return:
    """
    return "请求地址不存在"


@app.errorhandler(Exception)
def error(e):
    """
    其他异常处理
    :param e: 异常
    :return:
    """
    return str(e)


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=5000)

运行Flask服务

flask run
# 或者
python app.py

使用curl命令访问（略）
使用ApiFox/Postman接口测试工具访问（略）