Python - 连接并操作 MySQL

2022-08-06 23:13:18

接口测试什么情况下操作数据库？

比如说注册接口，同一个账号不能重复注册
流程性的操作，比如登录 - 新建学校 - 禁用学校，那么可以插入一个学校，然后再进行禁用操作
数据清理，测试完成后清理垃圾数据
修改状态，比如执行禁用动作的时候，state = 6，启用的时候state = 2，测试不同状态的时候，需要还原数据

time.sleep(20)
data = mydb.execute('DELETE FROM t_school_info WHERE f_school_name={}'.format(name))
print(data)

概述

主要讲解如何使用pymysql库进行MySQL的管理操作，
以及如何使用pymysql实现增删改查动作。

安装pymysql

pip install PyMySQL

常用对象及API

在 pymysql 中提供了 Connection 和 Cursor 对象来管理操作MySQL。

常用对象

Connection

代表一个与MySQL Server的socket连接，使用connect方法来创建一个连接实例。

Cursor
代表一个与MySQL数据库交互对象，使用Connection.Cursor()在当前socket连接上的交互对象。

常用API

Connection对象常用的API:

connect()   创建一个数据库连接实例  
begin()     开始一个事务  
close()     发送一个退出消息，并关闭连接  
commit()    提交修改至数据库  
cursor()    创建一个cursor(游标)实例  
ping()      检测服务器是否在运行  
rollback()  回滚当前事务  
select_db(db)   设置当前db  
show_warnings() 显示警告信息


Cursor对象常用API：

close()     关闭当前cursor
execute()   执行一个sql语句
executemany()   执行批量sql语句
fetchall()      取所有数据
fetchone()      取一条数据

栗子：

# -*- coding:utf-8 -*-

import pymysql
conn = pymysql.connect(
    # mysql 服务器 ip 地址
    host='192.168.28.128',
    # mysql 服务端口
    port=3306,
    # 访问 mysql 的用户名
    user='root',
    # 访问 mysql 的密码
    password='123123',
    # 默认连接到的数据库
    db='recruit_students'
)
# 使用 cursor 方法获取操作游标
cursor = conn.cursor()
sql = 'SELECT * FROM t_school_info'
# 使用 execute 方法执行sql查询
cursor.execute(sql)
data = cursor.fetchall()
print(data)

MySQL_utils数据库工具类封装

import pymysql
from warnings import filterwarnings

# 忽略Mysql告警信息
filterwarnings("ignore", category=pymysql.Warning)


class MysqlDb:
    def __init__(self):
        # 建立数据库连接
        self.conn = pymysql.connect(
            # mysql 服务器 ip 地址
            host='192.168.28.128',
            # mysql 服务端口
            port=3306,
            # 访问 mysql 的用户名
            user='root',
            # 访问 mysql 的密码
            password='123123',
            # 默认连接到的数据库
            db='recruit_students'
        )
        self.cur = self.conn.cursor()

    def __del__(self):
        # 关闭游标
        self.cur.close()
        # 关闭连接
        self.conn.close()

    def query(self, sql, state="all"):
        """
        查询
        :param sql:
        :param state: all是默认查询全部
        :return:
        """
        self.cur.execute(sql)

        if state == "all":
            # 查询全部
            data = self.cur.fetchall()
        else:
            # 查询单条
            data = self.cur.fetchone()
        return data

    def execute(self, sql):
        """
        更新、删除、新增
        :param sql:
        :return:
        """
        try:
            # 使用execute操作sql
            rows = self.cur.execute(sql)
            # 提交事务
            self.conn.commit()
            return rows
        except Exception as e:
            print("数据库操作异常 {0}".format(e))
            self.conn.rollback()


if __name__ == '__main__':
    mydb = MysqlDb()
    data = mydb.query('SELECT * FROM t_school_info')
    print(data)

码农公寓