Python接口测试实战3（上）- Python操做数据库

时间 2019-11-08

原文原文链接

课程目录

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

本节内容

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

前言

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

安装方法：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和获取结果api

查询操做
使用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能够完成各类业务操做
更改数据库有风险，操做需谨慎！！！

此为北京龙腾育才 Python高级自动化（接口测试部分）授课笔记
课程介绍
想要参加现场(北京)/网络课程的能够联系做者微信：lockingfree

高效学习，快速掌握Python自动化全部领域技能

同步快速解决各类问题

配套实战项目练习