2022年 11月 7日

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()
  1. import pymysql
  2. # 1. 建立连接
  3. conn = pymysql.connect(
  4.                 host='192.168.255.211',
  5.                 port=3306,
  6.                 user='root',
  7.                 passwd='123456',      # password也可以
  8.                 db='hellodb',
  9.                 charset='utf8')     # 如果查询有中文需要指定数据库编码             
  10. # 2. 从连接建立游标（有了游标才能操作数据库）
  11. cur = conn.cursor()
  12. # 3. 查询数据库（读）
  13. cur.execute("select * from hellodb where name='张三'")
  14. # 4. 获取查询结果
  15. result = cur.fetchall()
  16. print(result)
  17. # 3. 更改数据库（写）
  18. cur.execute("delete from hellodb where name='李四'")
  19. # 4. 提交更改
  20. conn.commit()  # 注意是用的conn不是cur
  21. # 5. 关闭游标及连接
  22. cur.close()
  23. conn.close()

1.1、查询操作

使用cur.execute()，执行数据库查询后无返回的是影响的行数，而非查询结果。

我们要使用cur.fetchone() / cur.fetchmany() / cur.fetchall()来获取查询结果

（1）cur.fetchone(): 获取一条数据（同时获取的数据会从结果集删除），返回元祖(‘张三’,’123456′)

（2）cur.fetchmany(3): 获取多条数据，返回嵌套元祖((‘张三’,’123456′),(‘李四’,’123456′),(“王五”,”123456″))

（3）cur.fetchall(): 获取所有数据，返回嵌套元祖，((‘张三’,’123456′),)（只有一条数据时）

获取完数据后，数据会从数据集中删除，再次获取获取不到，所以我们需要重复使用查询结果时，

需要将查询结果赋给某个变量

  1. cur.execute(select * from hellodb where name='张三')
  2. result = cur.fetchall()
  3. print(result)  # 结果： ('张三','123456')
  4. print(result)  # 结果： ('张三','123456')

2、封装数据库操作

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

  1. # 新建db.py
  2.  
  3. # 获取连接方法
  4. import pymysql
  5. # 获取连接方法
  6. def get_db_conn():
  7.     conn = pymysql.connect(
  8.                    host='192.168.255.211',
  9.                    port=3306,
  10.                    user='root',
  11.                    passwd='123456',  # password也可以
  12.                    db='hellodb',
  13.                    charset='utf8')  # 如果查询有中文需要指定数据库编码
  14.     return conn
  15. # 封装数据库查询操作
  16. def query_db(sql):
  17.     conn = get_db_conn()  # 获取连接
  18.     cur = conn.cursor()  # 建立游标
  19.     cur.execute(sql)  # 执行sql
  20.     result = cur.fetchall()  # 获取所有查询结果
  21.     cur.close()  # 关闭游标
  22.     conn.close()  # 关闭连接
  23.     return result  # 返回结果
  24.  
  25.  
  26. # 封装更改数据库操作
  27. def change_db(sql):
  28.     conn = get_db_conn()  # 获取连接
  29.     cur = conn.cursor()  # 建立游标
  30.     try:
  31.         cur.execute(sql)  # 执行sql
  32.         conn.commit()  # 提交更改
  33.     except Exception as e:
  34.         conn.rollback()  # 回滚
  35.     finally:
  36.         cur.close()  # 关闭游标
  37.         conn.close()  # 关闭连接
  38.  
  39.  
  40. # 封装常用数据库操作
  41. def check_user(name):
  42.     # 注意sql中''号嵌套的问题
  43.     sql = "select * from user where name = '{}'".format(name)
  44.     result = query_db(sql)
  45.     return True if result else False
  46.  
  47.  
  48. def add_user(name, password):
  49.     sql = "insert into user (name, passwd) values ('{}','{}')".format(name, password)
  50.     change_db(sql)
  51.  
  52.  
  53. def del_user(name):
  54.     sql = "delete from user where name='{}'".format(name)
  55.     change_db(sql)
  56.

相比用例中直接使用sql操作数据库，封装常用的数据库操作会更安全

  1. 调用方法(其他模块)
  1. from db import *
  2.  
  3. if check_user("张三"):
  4.     del_user("张三")

补充：另一种封装方法

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

  1. # 新建db1.py
  2. import pymysql
  3.  
  4. class DB:
  5.     def __init__(self):
  6.         self.conn = pymysql.connect(
  7.                    host='192.168.255.211',
  8.                    port=3306,
  9.                    user='root',
  10.                    passwd='123456',   # passwd 不是 password
  11.                     db='api_test')
  12.         self.cur = self.conn.cursor()
  13.         
  14.     def __del__(self): # 析构函数，实例删除时触发
  15.         self.cur.close()
  16.         self.conn.close()
  17.         
  18.     def query(self, sql):
  19.         self.cur.execute(sql)
  20.         return self.cur.fetchall()
  21.     
  22.     def exec(self, sql):
  23.         try:
  24.             self.cur.execute(sql)
  25.             self.conn.commit()
  26.         except Exception as e:
  27.             self.conn.rollback()
  28.             print(str(e))
  29.     
  30.     def check_user(self,name):
  31.         result = self.query("select * from user where name='{}'".format(name))
  32.         return True if result else False
  33.         
  34.     def del_user(self, name)
  35.         self.exec("delete from user where name='{}'".format(name))
  36.

使用方法：

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