使用Pymysql操作数据以及工具类封装

数据库操作应用场景

接口自动化测试数据库主要应用场景?

1.校验测试数据

接口发送请求后明确会对数据库中的某个字段进行修改,但,响应结果中无该字段数据时。

如:ihrm删除员工接口。 is_delete 字段,没有在响应结果中出现!需要借助数据库校验!

2.构造测试数据

测试数据使用一次就失效。如:ihrm 添加员工接口,使用的手机号!

测试前,无法保证测试数据是否存在。如:ihrm 查询员工接口,使用的员工id

数据库操作实现

图片[1] - 使用Pymysql操作数据以及工具类封装 - 三酷猫
PyMySQL操作数据库步骤
1. 导包 import pymysql
2. 创建连接。 conn = pymysql.connect(host,port, user, password, database, charset)
3. 获取游标。 cursor = conn.cursor()
4. 执行 SQL。 cursor.execute( ”sql语句“ )
   查询语句(select)
      处理结果集(提取数据 fetch*)
   增删改语句(insert、update、delete)
      成功:提交事务 conn.commit()
      失败:回滚事务 conn.rollback()
5. 关闭游标。cursor.close()
6. 关闭连接。conn.close()

1.事务

  • 事务,是关系型数据库(mysql)特有的概念。
  • 事务,可以看做一个虚拟的 容器,在容器中存放一系列的数据库操作,看做一个整体。内部的所有操作,要么都一次性全部成功,只要有一个失败,就全部失败!
图片[2] - 使用Pymysql操作数据以及工具类封装 - 三酷猫

事务操作:提交:conn.commit() ,回滚: conn.rollback()

2.游标

图片[3] - 使用Pymysql操作数据以及工具类封装 - 三酷猫
pymysql中游标的理解
cursor.fetchone()      #从结果集中,提取一行。
cursor.fetchmany(size) #从结果集中,提取size 行。
cursor.fetchall()      #提取所有结果集。
cursor.rownumber=0     #设置游标位置为0。

3.案例

"""
新增一条图书数据(id:5 title:西游记 pub_date:1986-01-01 )
insert into t_book(id, title, pub_date) values(5, '西游记', '1986-01-01');
1. 导包
2. 创建连接
3. 获取游标
4. 执行 insert 语句
5. 提交/回滚事务
6. 关闭游标
7. 关闭连接
"""

# 1. 导包
import pymysql

# 定义全局变量
conn = None
cursor = None

try:
    # 2. 创建连接
    conn = pymysql.connect(host="211.103.136.244", port=7061, user="student", password="iHRM_student_2021",
                           database="test_db", charset="utf8")
    # 3. 获取游标
    cursor = conn.cursor()

    # 4. 执行 insert 语句
    cursor.execute("insert into t_book(id, title, pub_date) values(175, '西游记', '1986-01-01');")

    # 查看 sql执行,影响多少行
    print("影响的行数:", conn.affected_rows())

    # 5. 提交事务
    conn.commit()

except Exception as err:
    print("插入数据错误:", str(err))
    # 回滚事务
    conn.rollback()

finally:
    # 6. 关闭游标
    cursor.close()
    # 7. 关闭连接
    conn.close()

数据库工具类封装

封装的目的

  • 将常用的数据库操作,封装到一个方法。 后续再操作数据库时,通过调用该方法来实现。
  • 提高代码的 复用性!
图片[4] - 使用Pymysql操作数据以及工具类封装 - 三酷猫

完整代码实现

import pymysql

# 封装数据库工具类
class DBUtil(object):
    # 添加类属性
    conn = None

    @classmethod
    def __get_conn(cls):
        # 判断 conn 是否为空,如果是,再创建
        if cls.conn is None:
            cls.conn = pymysql.connect(host="211.103.136.244", port=7061, user="student",
                                       password="iHRM_student_2021", database="test_db", charset="utf8")
        # 返回 非空连接
        return cls.conn

    @classmethod
    def __close_conn(cls):
        # 判断,conn 不为空,需要关闭。
        if cls.conn is not None:
            cls.conn.close()
            cls.conn = None

    # 常用方法:查询一条
    @classmethod
    def select_one(cls, sql):
        cursor = None
        res = None
        try:
            # 获取连接
            cls.conn = cls.__get_conn()

            # 获取游标
            cursor = cls.conn.cursor()

            # 执行 查询语句
            cursor.execute(sql)

            # 提取一条结果
            res = cursor.fetchone()

        except Exception as err:
            print("查询sql错误:", str(err))
        finally:
            # 关闭游标
            cursor.close()

            # 关闭连接
            cls.__close_conn()

            # 将查询sql执行的 结果,返回
            return res

    # 常用方法:增删改
    @classmethod
    def uid_db(cls, sql):
        cursor = None
        try:
            # 获取连接
            cls.conn = cls.__get_conn()

            # 获取游标
            cursor = cls.conn.cursor()

            # 执行 uid 语句
            cursor.execute(sql)
            print("影响的行数:", cls.conn.affected_rows())

            # 提交事务
            cls.conn.commit()

        except Exception as err:
            # 回滚事务
            cls.conn.rollback()
            print("增删改 SQL 执行失败:", str(err))

        finally:
            # 关闭游标
            cursor.close()
            # 关闭连接
            cls.__close_conn()


if __name__ == '__main__':
    res = DBUtil.select_one("select * from t_book;")
    print("查询结果为:", res)

    DBUtil.uid_db("update t_book set is_delete = 1 where id = 1111;")
© 版权声明
THE END
喜欢就支持一下吧
点赞8 分享