【Python+postman接口自动化测试】(12)Python 操作数据库

2021/11/11 2:13:32

本文主要是介绍【Python+postman接口自动化测试】(12)Python 操作数据库,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

Python 操作数据库  

在功能、接口测试中常常需要通过数据库的操作,来准备数据、检测环境及核对功能、接口的数据库操作是否正确。 在自动化测试中,就需要我们用代码连接数据库自动完成数据准备、环境检查及数据库断言的功能。 使用Python操作MySQL数据库这里我们需要用到三方库Pymysql
安装方法:pip install pymysql

1、数据库操作

1. 建立数据库连接 conn = pymysql.connect() 2. 从连接建立操作游标 cur = conn.cursor() 3. 使用游标执行sql(读/写) cur.execute(sql) 4. 获取结果(读)/ 提交更改(写) cur.fetchall() / conn.commit() 5. 关闭游标及连接 cur.close();conn.close()
import pymysql

# 建立数据库连接
conn = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    password='111111',  # passwd也可以
    db='school',
    charset='utf8')    # 如果查询有中文需要指定数据库编码

# 从连接建立游标,有了游标才能操作数据库
cur = conn.cursor()

# 更改数据库
cur.execute("INSERT INTO student VALUES ( 777, 'hello.殷', '男', 1990, '历史', '湖南宁乡' )")

# 查询数据库
cur.execute("select * from student where name='hello.殷'")

# 获取查询结果
result = cur.fetchall()
print(result)
# 提交更改
conn.commit()

 执行结果:

C:\Users\yzp\AppData\Local\Programs\Python\Python37\python.exe D:/00test/RFTEST/mysql.py
((777, 'hello.殷', '男', 1990, '历史', '湖南宁乡'),)

Process finished with exit code 0
什么是游标? 游标类似文件句柄,可以逐条的访问数据库执行结果集。pymysql中只能通过游标来执行sql和获取结果

2、查询操作

 使用cur.execute(), 执行数据库查询后无返回的是影响的行数,而非查询结果。我们要使用cur.fetchone()/cur.fetchmany()/cur.fetchall()来获取查询结果。

  • cur.fetchone(): 获取一条数据(同时获取的数据会从结果集删除),返回元祖('张三','123456')
  • cur.fetchmany(3): 获取多条数据,返回嵌套元祖(('张三','123456'),('李四','123456'),("王五","123456"))
  • cur.fetchall(): 获取所有数据,返回嵌套元祖,(('张三','123456'),)(只有一条数据时)
注意: 获取完数据后,数据会从数据集中删除,再次获取获取不到,如:
cur.execute(select * from user where name='张三')
print(cur.fetchone()) # 结果: ('张三','123456')
print(cur.fetchone()) # 结果:None
print(cur.fetchall()) # 结果:()

 所以我们需要重复使用查询结果时,需要将查询结果赋给某个变量

cur.execute(select * from user where name='张三')
result = cur.fetchall()
print(result) # 结果: ('张三','123456')
print(result) # 结果: ('张三','123456')

3、修改操作

执行修改数据库的操作后不立即生效,使用连接conn.commit()提交后才生效,支持事物及回滚
try:
  cur.execute("insert into user (name,password) values ('张三', '123456')")
  cur.execute("insert into user (name, passwd) values ('李四'), '123456'") # 此处sql出错
  conn.commit() # 使用连接提交所有更改
except Exception as e:
  conn.rollback() # 回滚所有更改(注意用的是conn)
  print(str(e))

 4、封装数据库操作

由于经常要使用到数据库操作,建议将所有数据库操作封装成公用的数据库模块

4.1 新建db.py, 代码如下:

import pymysql


class DB:
    def __init__(self):
        self.conn = pymysql.connect(
            host='127.0.0.1',
            port=3306,
            user='root',
            password='111111',
            db='school'
        )
        self.cur = self.conn.cursor()

    def __del__(self):  # 析构函数,实例删除时触发
        self.cur.close()
        self.conn.close()

    def query(self, sql):  # 查询函数
        self.cur.execute(sql)
        return self.cur.fetchall()

    def exc(self, sql):  # 执行函数
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            print(str(e))

    def check(self, name):
        result = self.query("select * from student where name='{}'".format(name))
        return True if result else False

    def del_user(self, name):
        self.exc("delete from student where name='{}'".format(name))

使用方法:

from db import * if check_user("张三"):   del_user("张三")

4.2 **补充:**另一种封装方法

由于上面这种封装方法,每一次查询都会建立一次数据库连接,效率较低,也可以采用下面面向对象的封装方法,新建db2.py
import pymysql


class DB:
    def __init__(self):
        self.conn = pymysql.connect(host='127.0.0.1',
                                    port=3306,
                                    user='root',
                                    passwd='111111',  # passwd 不是 password
                                    db='scholl')
        self.cur = self.conn.cursor()

    def __del__(self):  # 析构函数,实例删除时触发
        self.cur.close()
        self.conn.close()

    def query(self, sql):
        self.cur.execute(sql)
        return self.cur.fetchall()

    def exec(self, sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            print(str(e))

    def check_user(self, name):
        result = self.query("select * from user where name='{}'".format(name))
        return True if result else False

    def del_user(self, name):
        self.exec("del from user where name='{}'".format(name))

使用方法:

from db2 import DB: 
db = DB() # 实例化一个数据库操作对象
if db.check_user("张三"): 
  db.del_user("张三")

 

 5、后言

  • 数据库连接信息建议写到配置文件中,从配置文件中读取
  • sql语句建议先在手工测试一下没有语法问题再进行封装
  • 通过封装各种sql可以完成各种业务操作
  • 更改数据库有风险,操作需谨慎!!!


这篇关于【Python+postman接口自动化测试】(12)Python 操作数据库的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程