Python web 开发 flask 实践

1、前言

前文已经介绍了很多关于 python 的算法和脚本的写法，在本文将开启python的 web 的开发，和java 类似的，对于 web 开发也需要引入框架，对于 python 的 web 开发来说常见的有 flask 和 django 两种，在本文中将要介绍的是 flask 框架，相对来说起步简单一点儿。

2、环境准备

在开发之前，需要先安装 flask 的依赖，如下所示：

pip install flask

案例中的项目使用的环境是 python3.8, 不同于 java, python 的环境会因为依赖产生各种各样的问题，为了保证环境的一致性，需要保证依赖的版本一致。那么如何保证环境和版本的一致呢？

# 通常采用 pip freeze 生成当前环境的所有依赖
pip freeze > requirements.txt
# 采用 pip install 来安装文件中的所有依赖
pip install -r requirements.txt
# 如果只安装本项目的依赖，则需要安装 pip install pipreqs
# 通过以下命令来生成项目依赖的版本
pipreqs . --encoding=utf8 --force

3、项目结构

对于web项目开发，也可以根据 mvc 的原则来划分项目，如下图所示:

model 模块用来存放数据库表对应的实体
script 模块用来存放数据库脚本和sh 脚本
static 是项目的静态文件，这里使用的是前端后端不分离的模式，所以有静态模块和html末班
templates html 模版文件
web 即存放控制器的模块
business 存放业务模块
app 项目启动的入口
utils 即项目木启动所需要的工具类

对于 web 项目的创建，可以使用 pycharm 中的新建项目模块完成项目的创建和基础配置，也可以手工创建文件的方式来创建项目。

4、项目app

flask 的入口文件为 app.py，基础的配置以及依赖都在其中设置。这里通常配置 web 的访问和数据库的配置信息，还有项目的启动设置。

4.1 蓝图 Blueprint

通常情况下页面访问配置为 @app.route 即可访问，但是需要将页面路由的配置设置在 app.py 中，如果业务复杂或者路由过多的情况，就不太合适。通常在开发中可以使用蓝图来处理这类问题。

# 引入蓝图
from flask import Blueprint
# 注册蓝图，设置蓝图的名称，以及路径，url_prefix即访问前缀, 类似与spring controller 类上标注的 requestmapping。Login 为蓝图的名称，要唯一。
login_blue_print = Blueprint('Login', __name__, url_prefix="/login")

# 在使用蓝图时，需要使用的注解为 @login_blue_print.route 而非 app.route

4.2 数据库配置 sqlalchemy

对于数据库的配置，可以通过配置文件的方式来设置，具体的配置如下图所示：

# 配置方式
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True
# 配置文件的方式
app.config.from_object(config)

对于数据库的操作，可以使用 pymysql 的方式来操作数据库，对于脚本来说是合适的，但是对于 web 开发来说，还是需要采用面向对象的方式，这里使用的 orm 框架是 SQLAlchemy , 对于 django 框架来说也是同样的。

数据库实体类需要集成db.Model, 在实体类中可以设置数据库的表名称，字段名称，类型，默认值，注释，主键，索引，约束键等信息。数据库采用的是下划线的命名方式，对应的实体是驼峰命名。

class User(db.Model):

    __tablename__ = "tb_user"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment='主键')
    userId = db.Column("user_id", db.String(32), default="", unique=True, comment='用户Id')
    username = db.Column(db.String(32), default="", comment='用户')
    email = db.Column(db.String(32), default="", nullable=False, comment='邮箱')
    cellphone = db.Column(db.String(32), default="", comment='手机号')
    status = db.Column(db.String(32), default="",  comment='状态')
    seq = db.Column(db.Integer, default="", comment='seq')
    createTime = db.Column("create_time", db.DateTime, default=datetime.now, comment='创建时间')
    updateTime = db.Column("update_time", db.DateTime, default=datetime.now, comment='更新时间')

4.3 热加载

在开发模式下，需要开启热加载的模式，方便开发的调试，通过以下的方式可以实现热加载。

5、数据库操作

数据库的操作是开展业务的核心，简单的说就是如何在 python 中实现数据的增删改查，sqlalchemy 这个 orm 框架如何实现数据的增删改查呢?

5.1 数据新增

对于数据新增的场景，需要先创建对象，然后需要将对象加入的当前会话中，并提交修改即可。

user = User()
user.username= "小明"
user.age = 34
# 单条保存
db.session.add(user)
# 多条保存
# db.session.add_all([user, user2, user3])
# 事务失败会自动回滚
db.session.commit()

5.2 数据查询

数据的查询种类比较多，具体如下所示

# 返回列表, 元素为模型对象
usr = User.query.all()
# 根据条件查询数据的数量
cnt = User.query.count()
print("cnt ", cnt)
# 根据查询第一条数据，根据年龄age和 userId 排序，默认排序为顺序排 desc
fst = User.query.order_by(User.age.asc(), User.userId.desc()).first()
print("user ", autils.trans_item(fst))
res_list = autils.trans_items(usr)
# 查询id = 10 的用户数据
us = User.query.get(10)

# 等值过滤器 关键字实参设置字段值，返回查询的对象
# BaseQuery对象可以续接其他过滤器/执行器  如 all/count/first等
User.query.filter_by(id=4).all()
# 复杂过滤器，参数为比较运算/函数引用等
User.query.filter(User.id == 4).first()

# 查询名字进行查询，字符串结束，开始，包含，最常用的还是 like 模糊查询
User.query.filter(User.username.endswith("李")).all()
User.query.filter(User.username.startswith("明")).all()
User.query.filter(User.username.contains("晓")).all()
User.query.filter(User.username.like("赵")).all()  # 模糊查询
# 根据名称和邮箱进行查询，且的关系
User.query.filter(and_(User.username.startswith('li'), User.email.startswith('li'))).all()
# 或的关系
User.query.filter(or_(User.age == 25, User.email.endswith("itheima.com"))).all()
# 根据id list 进行查询数据
User.query.filter(User.id.in_([1, 3, 5, 7, 9])).all()
# 查询数据 offset limit 结合使用
User.query.order_by(User.age).offset(1).limit(4).all()

# 分页查询,查询第2页的数据，每页10个paginate(页码, 每页条数)
pn = User.query.paginate(page=2, per_page=10)

5.3 数据修改

对于数据的修改，需要先查询数据，然后修改完成之后进行保存即可。数据的修改有两种模式, 具体如下所示

# 1 执行查询语句, 获取目标模型对象
user = User.query.filter(User.id == 100).first()
# 2 对模型对象的属性进行赋值 (更新数据)
user.age = user.age + 1
# 3 提交会话
db.session.commit()

# 这里的修改传入的是一个字典，需要修改多个需要传入多个值即可。
User.query.filter(User.id = 1).update({'age': User.age - 1})
# 提交会话
db.session.commit()

5.4 数据删除

对于数据的删除，可以先查询后删除，也可以基于过滤条件的删除，和数据修改类似，具体如下所示：

# 1 执行查询语句, 获取目标模型对象
user = User.query.filter(User.id == 100).first()
# 2 删除数据
db.session.delete(user)
# 3 提交会话
db.session.commit()

# 这里可以根据多个过滤条件进行删除数据
User.query.filter(User.id = 1).delete()
# 提交会话
db.session.commit()

6、前端模块

这里使用的是前后端不分离的形式，如果使用前后端分离的方式，需要配置跨域：

from flask_cors import CORS
# 配置跨域 cors
cors = CORS(app, supports_credentials=True, resources=r'/*')

如果使用前后端不分离的方式，和 spring 的 model 类似，需要将参数传输到前端用于 html 的数据渲染。这里采用的是 flask 的 render_template，第一个参数是前端模版文件的名称，之后是在前端需要渲染的参数名以及参数值。

在前端渲染时，使用的方式和 jsp 类似，不过采用的是 {{}} 的方式来渲染参数，对于 for 循环和 if 条件判断也有类似的语法。

7、数据库连接池

前面讲述了如何使用 sqlalchemy 这个 orm 框架来查询数据，返回的对象需要经过转换才能返回到前端，在脚本中可以通过 dbutils.pooled_db 对 pymysql 进行包装来封装数据的操作。具体的操作如下所示：

import pymysql
from dbutils.pooled_db import PooledDB
# 声明数据库连接池
class DbPool(object):
    # 类变量数据库连接池
    __pool = None
    def __init__(self, db_config):
        # 构造函数，创建数据库连接、游标
        self.coon = DbPool.get_mysql_conn(db_config)
        self.cur = self.coon.cursor(cursor=pymysql.cursors.DictCursor)

    # 数据库连接池连接
    @staticmethod
    def get_mysql_conn(db_config):
        # 声明为全局变量
        global __pool
        if DbPool.__pool is None:
            __pool = PooledDB(creator=pymysql, mincached=1, maxcached=20, host=db_config['host'],
                              user=db_config['user'], passwd=db_config['passwd'], db=db_config['db'],
                              port=db_config['port'], charset=db_config['charset'])
        return __pool.connection()

    # 数据插入\更新\删除sql
    def op_update(self, sql, param):
        print('op_insert', sql, param)
        insert_num = self.cur.execute(sql, param)
        # commit 请求 
        self.coon.commit()
        return insert_num

    # 数据查询
    def op_query(self, sql, parm):
        print('op_select', sql, parm)
        self.cur.execute(sql, parm)  # 执行sql
        select_res = self.cur.fetchall()  # 返回结果为字典
        return select_res

    # 批量更新数据
    def op_update_list(self, list):
        sum = 0
        try:
            for tuple in list:
                sql, parm = tuple
                insert_num = self.cur.execute(sql, parm)
                if (insert_num == 0):
                    self.coon.rollback()
                    break
                sum += insert_num
            self.coon.commit()
        except:
            print('事务回滚')
            self.coon.rollback()
        return sum
        # 释放资源

    # 释放资源
    def dispose(self):
        self.coon.close()
        self.cur.close()
if __name__ == '__main__':
    # 数据库连接配置 
    db_config = {
        "host": 'localhost',
        "user": 'root',
        "passwd": '123456',
        "db": 'account',
        "port": 3306,
        "charset": 'utf8mb4'
    }

    db = DbPool(db_config)
    dt_list = db.op_query("select * from user where userid > %s order by userid desc limit 1", (23))
    for nd in dt_list:
        print(nd)

    res = db.op_update("update user set register_date = %s where userid > %s and userid < %s", ("2024-04-19 16:42:50", 17702, 17705))
    print("res is ", res)