python+request+pymysql+pytest数据驱动

乎语百科 2022-10-07 13:20:17 342 0

一.pymysql简单使用

1.安装mysql

下载地址：https://www.mysql.com/，安装教程这里不做介绍了，网上一大推。

2.安装pymysql库

在Terminal终端输入：pip install pymysql

3.pymysql操作数据库

1）导入pymysql库，创建数据库连接

conn = pymysql.connect(host="127.0.0.1",  # 本地服务ip                       user="root",  # 用户名                       password="root",  # 密码                       database="pymysql",  # 库                       port=3306,  # 端口号                       autocommit=False,  # 是否自动提交                       charset="utf8")  # 设置utf编码

2）创建游标

execute()执行sql语句

cur = conn.cursor()  # 创建游标
sql = "create table student(id int(10),name varchar(10))"
cur.execute(sql)  # 执行sql,执行后返回的是影响的条数

创建游标执行sql语句后fetchall()默认返回元组类型的数据，我们要想让其返回dict类型的，可加入相应参数

cur = conn.cursor(pymysql.cursors.DictCursor)

executemany()执行sql语句

a_list = [(1, "李明"), (2, "李祥")]
sql = "insert into student (id,name ) values (%s,%s)"  # %s占位符，通用的
cur_list = cur.executemany(sql, a_list)  # 批量执行sql

executemany(self, query, args)：为执行批量语句时使用的，query为执行的sql语句，args为传入的数据，列表中嵌套元组类型。

execute()也可批量执行语句，只是执行效率没executemany()快，因此执行单条语句时推荐使用execte()，批量时推荐使用exectemany()。

3）返回查询数据的操作

cur.fetchone() ：返回查询第一条的结果集

cur.fetchall() ：返回所有的查询结果集，元组类型

cur.fetchmany(): 默认返回第一条结果集，可手动在括号中填写返回条数

cur.rowcount：返回影响的条数。注意这里是属性，不是方法

4）commit的使用

我们在操作数据库时，进行增删改后需求提交，否则的话数据库对应的数据是不会发生变化的

conn.commit()

若的创建数据库连接的时候autocommit设置为True，自动提交后。此步骤可省略

5）执行完毕后，关闭游标和数据库连接

cur.close()
conn.close()

二.数据驱动

1）我们以获取企业微信token为例，首先先从数据库获取相应的数据

def pymysql_01():
    # 创建数据库连接
    conn = pymysql.connect(host="127.0.0.1",  # 本地服务ip
                           user="root",  # 用户名
                           password="root",  # 密码
                           database="li",  # 库
                           port=3306,  # 端口号
                           autocommit=False,  # 是否自动提交
                           charset="utf8")  # 设置utf编码
    cur = conn.cursor(pymysql.cursors.DictCursor)  # 创建游标 pymysql.cursors.DictCursor为查询时返回dict类型
    sql = "select * from pymysql"
    cur.execute(sql)
    list_id = []
    for i in cur.fetchall():
        list_id.append(i)
    cur.close()
    conn.close()
    return list_id

2）@pytest.mark.parametrize()实现数据驱动

@pytest.mark.parametrize("param", pymysql_01())
def test_weixin(param):
    url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
    data = requests.get(url, params=param)
    assert data.json()["errcode"] == 0

3）运行结果当然也是完美的

文章来源：https://www.cnblogs.com/lihongtaoya/ ，请勿转载

标签： # MySQL # python