PyMySQL包
掌握 PyMySQ L对数据库实现增、删、改、查
数据库工具类封装
数据库操作应用场景
- 校验 测试数据
- 接口发送请求后明确会对数据库中的某个字段进行修改，但，响应结果中无该字段数据时。
- 如：ihrm 删除员工接口。 数据表中的is_delete 由0变1，但没有在响应结果中出现！ 需要借助数据库校验！
- 构造 测试数据
- 测试数据使用一次就失效。
- 如：注册接口，新的手机号只能使用一次！
- 测试前，无法保证测试数据是否存在。
- 如：查询员工接口，使用的 员工id 需要先从数据库获取。
简介与安装
pymysql 是从Python连接到MySQL数据库服务器的接口，其官方文档为：https://pymysql.readthedocs.io/en/latest/
安装命令：pip install pymysql
基本使用
1. 导包 import pymysql
2. 创建连接。 conn = pymysql.connect(host，port, user, password, database, charset)
3. 获取游标。 cursor = conn.cursor()
4. 执行 SQL。 cursor.execute( ”sql语句“ )
- 查询语句（select）
-
- 处理结果集（提取数据 fetchone()、fetchmany(size)、fetchall()）
- 增删改语句（insert、update、delete）
-
- 成功：提交事务 conn.commit()
-
- 失败：回滚事务 conn.rollback()
5. 关闭游标。cursor.close()
6. 关闭连接。conn.close()
案例：查询数据表
查询数据库mydb的表stu
# 导包 import pymysql # 1、连接数据库 conn = pymysql.connect(host="localhost", port=3306, user="root", password="root", database="mydb", charset="utf8") # 2、创建游标 cursor = conn.cursor() # 3、执行select语句 cursor.execute("select * from account;") # 3.1.获取1条记录，从下标0开始 cursor.rownumber = 0 # 修改游标位置：回零 resp_fetchone = cursor.fetchone() print("resp_fetchone: ", resp_fetchone) # 3.2.获取2条记录，从下标1开始 cursor.rownumber = 1 # 修改游标位置：下标1 resp_fetchmany = cursor.fetchmany(2) print("resp_fetchmany: ", resp_fetchmany) # 3.3.获取所有记录，从下标0开始 cursor.rownumber = 0 # 修改游标位置：回零 resp_fetchall = cursor.fetchall() print("resp_fetchall: ", resp_fetchall) # 4、关闭游标 cursor.close() # 5、关闭连接 conn.close()
案例：增、删、改数据表
从以上案例可以看出，代码冗余，所以，可以采用封装
封装工具类
连接数据库__get_conn(cls)，关闭连接__close_conn(cls)，查询select_one(cls,sql)，增删改uid_db(cls,sql)
import pymysql # 该类用于封装数据库操作：创建连接、关闭连接、查询、更新、修改、删除 class DBUtil(object): conn = None # 连接数据库------------------ @classmethod def __get_conn(cls): # 判断 conn 是否为空， 如果是，再创建 if cls.conn is None: cls.conn = pymysql.connect(host="localhost", port=3306, user="root", password="root", database="mydb", charset="utf8") return cls.conn # 关闭连接------------------ @classmethod def __close_conn(cls): if cls.conn is not None: cls.conn.close() cls.conn = None # 常用方法fetchone()：查询一条------------- @classmethod def select_one(cls, sql): cursor = None resp = None try: # 连接数据库 cls.conn = cls.__get_conn() # 创建游标 cursor = cls.conn.cursor() # 执行sql语句 cursor.execute(sql) # 只提取一条记录 resp = cursor.fetchone() # resp = cursor.fetchmany(size) # 提取size条 # resp = cursor.fetchall() # 提取全部 except Exception as err: print("查询错误", str(err)) finally: # 关闭游标 cursor.close() # 关闭连接 cls.__close_conn() # 返回查询结果 return resp # 增删改------------------- @classmethod def uid_db(cls, sql): cursor = None resp = None try: # 连接数据库 cls.conn = cls.__get_conn() # 创建游标 cursor = cls.conn.cursor() # 执行sql语句 cursor.execute(sql) print("影响行数：", cls.conn.affected_rows()) # 提交事务 cls.conn.commit() except Exception as err: print("操作错误:", str(err)) # 回滚事务 cls.conn.rollback() finally: # 关闭游标 cursor.close() # 关闭连接 cls.__close_conn()
调用上面工具类的方法来实现操作
import pymysql from pymysql_tools import DBUtil # 用封装类里的方法，只需sql传参 # 查询select DBUtil.select_one("select * from stu where name='小明';") # 插入insert DBUtil.uid_db("insert into stu(name,sex,age,classRoom_id) values('张三','男','18',1);") DBUtil.uid_db("insert into stu(name,sex,age,classRoom_id) values('李四','男','18',2);") # 更新update DBUtil().uid_db("update stu set age='20' where name='李四';") # 删除delete DBUtil.uid_db("delete from stu where name=‘王五’;")
