🎉 此分享将会是一个系列分享，将使用最新的 Acro Vue 前端框架，和 Python Flask 服务从 0-1 手把手实现一个测试需求平台，内容将囊括基础、扩展和实战，由浅入深带你实现测试开发岗位中平台工具技术能力入门和提升。
作者：大奇 MegaQi | Info 签约作者 ，专注测试开发技术实战干货分享，欢迎访问主页长期关注。
本篇需要提前准备的环境和开发内容：
准备数据库，Mysql5.7+本地或云服务均可
实现后端接口服务的数据库操作
1. 产品数据持久化
在项目管理中，真正的数据需要持久化操作的，这里必然就离不开数据库，本项目使用的 Mysql 数据库，但不会过多的讲解 SQL 的内容，只会重点讲解后端服务中 Python 对于数据库的操作相关知识点。
1.1 数据库和产品表初始化
使用数据库 IDE 工具链接 mysql 数据库，并创建一个数据库 TPMStore 和一个 Products 表，字段分别如下
使用 Navicat 可视化创建，或查看笔者大奇之前分享过的一个好用的开源 Beekeeper 工具
这里给出 SQL 语句方便进行表格创建，顺便添加两条正式测试数据
SET NAMES utf8mb4;SET FOREIGN_KEY_CHECKS = 0;
-- ------------------------------ Table structure for products-- ----------------------------DROP TABLE IF EXISTS `products`;CREATE TABLE `products` ( `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号自增', `keyCode` varchar(200) NOT NULL COMMENT '项目唯一编号', `title` varchar(200) NOT NULL COMMENT '中文项目名', `desc` varchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '' COMMENT '描述', `operator` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '操作者', `update` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '操作时间', PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='项目产品表';
-- ------------------------------ Records of products-- ----------------------------BEGIN;INSERT INTO `products` VALUES (1, 'data', '数据大盘', '内部一个数据技术分析的项目，用于分析各种数据聚合平台', 'daqi', '2021-07-17 20:38:37');INSERT INTO `products` VALUES (2, 'payment', '收银台', '支付聚合收银台', 'lili', '2021-07-17 20:40:29');COMMIT;
SET FOREIGN_KEY_CHECKS = 1;
复制代码
1.2 产品查询接口改造
之前的接口 /api/product/list 是硬编码返回，我们这里需要改造成通过数据库查询获得。
python 实现 mysql 的数据的方式目前支持度较好的有：
以上 github star 数据统计于 2022/07/10
综合使用度和后续可能使用 ORM(对象关系映射)优化，本项目选择 PyMySQL
# 安装依赖包python3 -m pip install PyMySQL
复制代码
然后主要就是引入包，实现数据库的连接和查询操作
# -*- coding:utf-8 -*-from flask import Blueprintimport pymysql.cursors
app_product = Blueprint("app_product", __name__)
# 使用用户名密码创建数据库链接connection = pymysql.connect(host='localhost', # 数据库IP地址或链接域名 user='root', # 设置的具有增改查权限的用户 password='******', # 用户对应的密码 database='TPMStore',# 数据表 charset='utf8mb4', # 字符编码 cursorclass=pymysql.cursors.DictCursor) # 结果作为字典返回游标
@app_product.route("/api/product/list",methods=['GET'])def product_list(): # 使用python的with..as控制流语句（相当于简化的try except finally） with connection.cursor() as cursor: # 查询产品信息表-按更新时间新旧排序 sql = "SELECT * FROM `Products` ORDER BY `Update` DESC" cursor.execute(sql) data = cursor.fetchall()
# 按返回模版格式进行json结果返回 resp_data = { "code": 20000, "data": data } return resp_data
复制代码
对改造的后的接口用 Postman 做个验证测试，从下图可以看到数据已经是从 product 表里查出来的。
2. PyMySQL 使用
PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库。
2.1 依赖安装
首先关于安装依赖，上边已经给列出通过$ python3 -m pip install PyMySQL 命令，这里需要特别强调一下，如果你使用的是 Mysql 8.x 数据库服务，由于高版本改变了密码加密方式，所以必须安装额外的依赖
# 不兼容加密方式连接报错如下# RuntimeError: 'cryptography' package is required for sha256_password or caching_sha2_password auth methods$ python3 -m pip install PyMySQL[rsa]
复制代码
另外还可以通过改 mysql 的加密方式为 mysql_native_password，笔者本地版本8.0.25为了方便已经修改了，查询的方法见截图：
2.2 数据库连接
连接实例的是在代码中 import pymysql 后通过.connet(...)方法创建，扒一下源码可以看到有很到参数
这里捡一些基本的和可能用到的参数做下解释说明
host 数据库地址 IP 或者域名
user 数据库分配的账号
password 数据库分配的密码
database 指定数据库, 可以为 None 后续使用时候进行切换
port 数据库端口号，默认 3306
read_timeout 读操作超时时间
write_timeout 写操作超时时间
charset 字符编码中文建议设置 utf8 等
cursorclass 设置返回的数据类型，默认查询返回的数据是 tuples 元组，一般我们数据多数以 table 的形式呈现，所以建议设置成cursorclass=MySQLdb.cursors.DictCursor字典类型 h
connect_timeout 连接超时时间
还有其他如 ssl 加密相关、连接数设置等，实际项目中按需查阅参数配置。
另外一点在创建 db 对象后，其实就可以通过游标创建对应的数据库以及切换对应的库。
# 执行创建数据库cursor.execute("CREATE DATABASE QiDBTest character SET utf8mb4;")
# 切库或使用 db.select_db("数据库名")cursor.execute("use QiDBTest;")
# #可以通过以下语句进行创建查询# cursor.execute("SHOW DATABASES;")# print(cursor.fetchall())
复制代码
2.3 数据库表操作
创建数据库连接对象，然后再创建一个游标对象cursor，通过cursor.excute() 执行对应的语句，就可以进行表相关、数据相关操作，其实 excute 的操作，你完全可以被看做使用任何一个数据库 IDE 工具，打开了一个查询面板来执行对应的 SQL 语句
表创建和数据查询 均通过执行对应的 SQL 语句实现，其中查询结果还需要通过cursor.fetchall()获取，对应的还有两个常用的
import pymysql # 创建数据库连接db = pymysql.connect(host='localhost', user='mrzcode', password='mrzcode', database='QiDBTest') # 使用cursor()方法创建一个游标对象 cursorcursor = db.cursor()
# 使用execute()方法执行 SQL，如果表存在则删除cursor.execute("DROP TABLE IF EXISTS QiTableDemo")
# 创建表语句sqlCreateTable = """ CREATE TABLE `QiTableDemo` ( `id` bigint NOT NULL AUTO_INCREMENT COMMENT '自增ID', `name` varchar(100) NOT NULL COMMENT '名称', `desc` varchar(500) COMMENT '描述', PRIMARY KEY (`id`) )"""
# 执行SQL语句创建表cursor.execute(sqlCreateTable)
# 验证查询下表情况cursor.execute("SHOW TABLES;")print(cursor.fetchall()) # (('qitabledemo',),) 默认元组查询表列表
# 查询QiTableDemo表所有数据sqlSelect = "SELECT * FROM QiTableDemo;"
# 执行表查询语句cursor.execute(sqlSelect)print(cursor.fetchall()) # () 新表返回一个空的tuples
复制代码
表数据增删改 额外在 execute 基础上进行db.commit()提交，如果不提交连接关闭后这些数据修改是不生效的。
import pymysql
# 创建数据库连接db = pymysql.connect(host='127.0.0.1', user='root', password='rox@123', database='QiDBTest', cursorclass=pymysql.cursors.DictCursor)
# 使用 cursor() 方法创建一个游标对象 cursorcursor = db.cursor()
sqlInsert = '''INSERT INTO qitabledemo(`name`,`desc`) VALUES ('插入测试名称', '插入测试描述');'''
# 执行表查询语句cursor.execute(sqlInsert)
# 对执行提交，这里可以尝试注释掉验证不进行提交数据能否插入db.commit()
# 查询数据是否正确插入cursor.execute("select * from qitabledemo;")print(cursor.fetchall()) # [{'id': 6, 'name': '插入测试名称', 'desc': '插入测试描述'}]
# 关闭数据库连接db.close()
复制代码
上边这种 sql 语句是一个字符串形式，但实际在代码逻辑处理中值一般都是通过变量传递的，所以通过以下两种方式动态赋值
# 方式一：占位拼接字符串sqlMethod1 = "INSERT INTO qitabledemo(`name`,`desc`) VALUES ('%s', '%s')" % ("占位名称","占位描述")cursor.execute(sqlMethod1)
# 方式二：通过execute传参sqlMethod2 = "INSERT INTO qitabledemo(`name`,`desc`) VALUES ('%s', '%s')"cursor.execute(sqlMethod1，(变量1,变量2))
复制代码
剩下关于更新、删除的操作同样，只是按需替换成对应的语句，但强调的一点是不要忘记commit，以下给出我这边的例子和验证测试
2.4 事务和错误处理
关于事务机制 可以确保数据一致性，场景主要用于多逻辑交互时候其中操作错误，进行响应的回滚处理，避免产生脏数据，事务通常具有 4 个属性：原子性、一致性、隔离性、持久性。
对于支持事务的数据库， 在 Python 数据库编程中，当游标建立之时，就自动开始了一个隐形的数据库事务。
每一个方法都开始了一个新的事务。
sql = "数据库操作的增删改查语句"try: # 执行SQL语句 cursor.execute(sql) # 向数据库提交 db.commit()except: # 发生错误时回滚 db.rollback()
复制代码
关于错误 DB API 中定义了一些数据库操作的错误及异常（以下引用菜鸟教程），严谨的编程需要对不同的错误进行响应的处理。
2.6 新手操作指南
最后总结一下一般 Python 使用 PyMySQL 的编程步骤
引用模块库
创建连接对象 db=connect(...)
从连接对象获取游标 cursor=db.cursor()
准备 sql 语句并通过游标执行 cursor.execute(sql)
如果是非查询动作还需要 db.commit()
关闭数据库连接 db.close()
以上就是本篇的主要内容，重点讲解 Python 对 mysql 数据库的操作，并且开始就开门见山地做个了项目实战，相信这些内容掌握了，本系列项目中有关数据操作部分都会游刃有余。
如喜欢通过“点赞👍收藏❤️关注➕”鼓励作者大奇
文章中如有错误或疑问欢迎指正和交流。
评论