python 操做数据库 转载：https://www.jianshu.com/p/736ed1e724f0

课程目录

Python接口测试实战1（上）- 接口测试理论
Python接口测试实战1（下）- 接口测试工具的使用
Python接口测试实战2 - 使用Python发送请求
Python接口测试实战3（上）- Python操做数据库
Python接口测试实战3（下）- unittest测试框架
Python接口测试实战4（上） - 接口测试框架实战
Python接口测试实战4（下） - 框架完善：用例基类，用例标签，从新运行上次失败用例
Python接口测试实战5（上） - Git及Jenkins持续集成
Python接口测试实战5（下） - RESTful、Web Service及Mock Server

更多学习资料请加添加做者微信：lockingfree获取

本节内容

• 数据库操做
• 封装数据库操做

前言

在功能、接口测试中经常须要经过数据库的操做，来准备数据、检测环境及核对功能、接口的数据库操做是否正确。
在自动化测试中，就须要咱们用代码链接数据库自动完成数据准备、环境检查及数据库断言的功能。
使用Python操做MySQL数据库这里咱们须要用到三方库PyMySQl

安装方法：pip install pymysql

数据库操做

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 # 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)) 

封装数据库操做

因为常常要使用到数据库操做，建议将全部数据库操做封装成公用的数据库模块

1. 新建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操做数据库，封装经常使用的数据库操做会更安全

1. 调用方法(其余模块)

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://www.jianshu.com/p/736ed1e724f0 来源：简书 著做权归做者全部。商业转载请联系做者得到受权，非商业转载请注明出处。