unittest(7):Python接口自动化之pymysql数据库操作

在上一篇Python接口自动化测试系列文章：Python接口自动化之yaml配置文件，主要介绍主要介绍yaml语法、yaml存储数据，封装类读写yaml配置文件。

在自动化过程中，我们需要查询数据库，校验结果是否正确，比如充值完成之后，需要查询数据库，查看充值是否成功。

以下主要介绍，pymysql安装、操作流程、语法基础及封装操作数据库类。

一 pymysql介绍及安装

01 pymysql介绍

MySQL应该说是如今使用最为普遍的数据库了，没有之一，而Python作为最为流行的语言之一，自然少不了与MySQL打交道，其中PyMySQL就是使用最多的工具库。

PyMySQL是一个纯Python写的MySQL客户端，可以在CPython、PyPy、IronPython和Jython环境下运行；

PyMySQL的性能和MySQLdb几乎相当，如果对性能要求不是特别的强，使用PyMySQL将更加方便；

PyMySQL的使用方法和MySQLdb几乎一样；

二 pymysql流程及模块说明

01 pymysql操作流程

导入pymysql；
建立数据库连接：使用pymysql的connect()方法连接数据，返回连接对象；
使用连接对象创建游标对象（用于操作sql）;
准备写sql语句（select * from student）;
使用游标对象执行sql;
查询数据使用游标获取;
关闭游标（先）和数据库连接（后）。

02 pymysql模块说明

▌Connection对象

表示：conn=connect(参数列表)

作用：用于建立与数据库的连接；

创建对象：调用connect()方法；

参数列表：

host：连接的mysql主机，如本机是'localhost'；
port：连接的mysql主机的端口，默认是3306；
database：数据库的名称；
user：连接的用户名；
password：连接的密码；
charset：通信采用的编码方式，推荐使用utf8//不是utf-8；

▌对象的方法

对象方法如下：

close()：关闭连接；
commit()：提交；
cursor()：返回Cursor对象，用于执行sql语句并获得结果；
execute(operation [, parameters ])：执行语句，返回受影响的行数，主要用于执行insert、update、delete语句，也可以执行create、alter、drop等语句；
fetchone()：执行查询语句时，获取查询结果集的第一个行数据，返回一个元组；
fetchall()：执行查询时，获取结果集的所有行，一行构成一个元组，再将这些元组装入一个元组返回；
关于pymysql防注入，字符串拼接查询，容易造成注入，为了避免注入，使用pymysql提供的参数化语句;

02 语法总结

1.连接数据库，需要host、user、password、database、charset等信息；

2.操作数据库先创建游标；

3.执行指定的sql语句，如果涉及到增、删、改数据库必须要conn.commit()，提交事务

4.查询获取数据条数有三种方法fetchone、fetchmany、fetchall。

cursor.fetchone() ：默认获取查询结果的第一条数据；
cursor.fetchmany(2) ：获取查询结果的指定条数，比如获取2条数据；
cursor.fetchall() ：获取查询结果的所有数据；

5.需要注意的是，fetch获取的数据默认是元组，如果想要字典类型，

cursor=pymysql.cursors.DictCursor；

6.先关闭游标，后关闭数据库连接；

四 封装数据库类

01 封装说明

在实际项目中，很多地方都有用到数据库的操作，所以需要将数据库相关操作进行封装，方便其他模块调用。

如下，在common目录下，新建文件db_handler.py 用于封装数据库操作。

import pymysql

from pymysql1.yaml_handle import test_yaml


class db_connect(object):
    def __init__(self,host,port,username,pwd,database,charset):
        self.conn=pymysql.connect(host=host,port=port,user=username,password=pwd,database=database,charset=charset)
        self.course=self.conn.cursor()
    def query(self,sql,cishu=1):
        self.course.execute(sql)
        self.conn.commit()
        if cishu==1:
            return self.course.fetchone()
        else:
            return self.course.fetchall()
    def db_close(self):
        self.course.close()
        self.conn.close()
if __name__ == '__main__':
    ty=test_yaml('testYaml.yaml')
    db_dict=ty.read_yaml()
    host=db_dict['db']['host']
    port=db_dict['db']['port']
    username=db_dict['db']['username']
    pwd=db_dict['db']['pwd']
    database=db_dict['db']['database']
    charset=db_dict['db']['charset']
    db=db_connect(host=host,port=port,username=username,pwd=pwd,database=database,charset=charset)
    sql='SELECT sid FROM student WHERE student.sname="赵"'
    print(db.query(sql=sql, cishu=2)[0][0])
    db.db_close()
    # print(host,port,username,pwd,database,charset)

原文地址：https://www.cnblogs.com/lhTest/p/15006913.html