本节内容
- 数据库操作
- 封装数据库操作
前言
在功能、接口测试中常常需要通过数据库的操作，来准备数据、检测环境及核对功能、接口的数据库操作是否正确。
在自动化测试中，就需要我们用代码连接数据库自动完成数据准备、环境检查及数据库断言的功能。
使用Python操作MySQL数据库这里我们需要用到三方库PyMySQl
安装方法：
pip install pymysql
数据库操作#
- 建立数据库连接
conn = pymysql.connect() - 从连接建立操作游标
cur = conn.cursor() - 使用游标执行sql（读/写）
cur.execute(sql) - 获取结果（读）/ 提交更改（写）
cur.fetchall()/conn.commit() - 关闭游标及连接
cur.close();conn.close()
-
import pymysql
-
-
# 1. 建立连接
-
conn = pymysql.connect(host=
'127.0.0.1',
-
port=
3306,
-
user=
'root',
-
passwd=
'123456',
# password也可以
-
db=
'api_test',
-
charset=
'utf8')
# 如果查询有中文需要指定数据库编码
-
-
# 2. 从连接建立游标（有了游标才能操作数据库）
-
cur = conn.cursor()
-
-
# 3. 查询数据库（读）
-
cur.execute(
"select * from user where name='张三'")
-
-
# 4. 获取查询结果
-
result = cur.fetchall()
-
print(result)
-
-
# 3. 更改数据库（写）
-
cur.execute(
"delete from user where name='李四'")
-
-
# 4. 提交更改
-
conn.commit()
# 注意是用的conn不是cur
-
-
# 5. 关闭游标及连接
-
cur.close()
-
conn.close()
什么是游标？ 游标类似文件句柄，可以逐条的访问数据库执行结果集。pymysql中只能通过游标来执行sql和获取结果
查询操作
使用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')
修改操作
执行修改数据库的操作后不立即生效，使用连接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))
封装数据库操作
由于经常要使用到数据库操作，建议将所有数据库操作封装成公用的数据库模块
- 新建db.py, 代码如下：
-
import pymysql
-
-
-
# 获取连接方法
-
import pymysql
-
-
-
# 获取连接方法
-
def
get_db_conn():
-
conn = pymysql.connect(host=
'127.0.0.1',
-
port=
3306,
-
user=
'test',
-
passwd=
'123456',
-
db=
'api_test',
-
charset=
'utf8')
# 如果查询有中文，需要指定测试集编码
-
-
return conn
-
-
-
# 封装数据库查询操作
-
def
query_db(
sql):
-
conn = get_db_conn()
# 获取连接
-
cur = conn.cursor()
# 建立游标
-
cur.execute(sql)
# 执行sql
-
result = cur.fetchall()
# 获取所有查询结果
-
cur.close()
# 关闭游标
-
conn.close()
# 关闭连接
-
return result
# 返回结果
-
-
-
# 封装更改数据库操作
-
def
change_db(
sql):
-
conn = get_db_conn()
# 获取连接
-
cur = conn.cursor()
# 建立游标
-
try:
-
cur.execute(sql)
# 执行sql
-
conn.commit()
# 提交更改
-
except Exception
as e:
-
conn.rollback()
# 回滚
-
finally:
-
cur.close()
# 关闭游标
-
conn.close()
# 关闭连接
-
-
-
# 封装常用数据库操作
-
def
check_user(
name):
-
# 注意sql中''号嵌套的问题
-
sql =
"select * from user where name = '{}'".
format(name)
-
result = query_db(sql)
-
return
True
if result
else
False
-
-
-
def
add_user(
name, password):
-
sql =
"insert into user (name, passwd) values ('{}','{}')".
format(name, password)
-
change_db(sql)
-
-
-
def
del_user(
name):
-
sql =
"delete from user where name='{}'".
format(name)
-
change_db(sql)
相比用例中直接使用sql操作数据库，封装常用的数据库操作会更安全
- 调用方法(其他模块)
-
from db import *
-
-
if
check_user("张三"):
-
del_user(
"张三")
补充：另一种封装方法
由于上面这种封装方法，每一次查询都会建立一次数据库连接，效率较低，也可以采用下面面向对象的封装方法
db2.py
-
import pymysql
-
-
-
class
DB:
-
def
__init__(
self):
-
self.conn = pymysql.connect(host=
'127.0.0.1',
-
port=
3306,
-
user=
'root',
-
passwd=
'123456',
# passwd 不是 password
-
db=
'api_test')
-
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(
"delete from user where name='{}'".
format(name))
使用方法
-
from db2 import DB:
-
-
db =
DB() # 实例化一个数据库操作对象
-
if db.
check_user(
"张三"):
-
db.
del_user(
"张三")
后言
- 数据库连接信息建议写到配置文件中，从配置文件中读取
- sql语句建议先在手工测试一下没有语法问题再进行封装
- 通过封装各种sql可以完成各种业务操作
- 更改数据库有风险，操作需谨慎！！！
转载：https://blog.csdn.net/ZangKang1/article/details/128487732
查看评论