介面測試什麼情況下操作資料庫？

1. 比如說註冊介面，同一個賬號不能重複註冊
2. 流程性的操作，比如登入 - 新建學校 - 禁用學校，那麼可以插入一個學校，然後再進行禁用操作
3. 資料清理，測試完成後清理垃圾資料
4. 修改狀態，比如執行禁用動作的時候，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)