如何實(shí)現(xiàn)Python接口測(cè)試數(shù)據(jù)庫(kù)封裝

import pymysql
import json
 
 
class OperateMysql(object):
  def __init__(self):
    # 數(shù)據(jù)庫(kù)初始化連接
    self.connect_interface_testing = pymysql.connect(
      "localhost",
      "root",
      "123456",
      "test",
      charset='utf8mb4',
      cursorclass=pymysql.cursors.DictCursor
    )
 
    # 創(chuàng)建游標(biāo)操作數(shù)據(jù)庫(kù)
    self.cursor_interface_testing = self.connect_interface_testing.cursor()
 
  def select_first_data(self, sql):
    """
    查詢第一條數(shù)據(jù)
    """
    try:
      # 執(zhí)行 sql 語(yǔ)句
      self.cursor_interface_testing.execute(sql)
    except Exception as e:
      print("執(zhí)行sql異常:%s"%e)
    else:
      # 獲取查詢到的第一條數(shù)據(jù)
      first_data = self.cursor_interface_testing.fetchone()
      # print(first_data)
      # 將返回結(jié)果轉(zhuǎn)換成 str 數(shù)據(jù)格式，禁用acsii編碼
      first_data = json.dumps(first_data,ensure_ascii=False)
      # self.connect_interface_testing.close()
      return first_data
 
  def select_all_data(self,sql):
    """
    查詢結(jié)果集
    """
    try:
      self.cursor_interface_testing.execute(sql)
    except Exception as e:
      print("執(zhí)行sql異常:%s"%e)
    else:
      first_data = self.cursor_interface_testing.fetchall()
      first_data = json.dumps(first_data,ensure_ascii=False)
      # self.connect_interface_testing.close()
      return first_data
 
  def del_data(self,sql):
    """
    刪除數(shù)據(jù)
    """
    res = {}
    try:
      # 執(zhí)行SQL語(yǔ)句
      result = self.cursor_interface_testing.execute(sql)
      # print(result)
      if result != 0:
        # 提交修改
        self.connect_interface_testing.commit()
        res = {'刪除成功'}
      else:
        res = {'沒(méi)有要?jiǎng)h除的數(shù)據(jù)'}
    except:
      # 發(fā)生錯(cuò)誤時(shí)回滾
      self.connect_interface_testing.rollback()
      res = {'刪除失敗'}
    return res
 
  def update_data(self,sql):
    """
    修改數(shù)據(jù)
    """
    try:
      self.cursor_interface_testing.execute(sql)
      self.connect_interface_testing.commit()
      res = {'更新成功'}
    except Exception as e:
      self.connect_interface_testing.rollback()
      res = {'更新刪除'}
    return res
 
  def insert_data(self,sql,data):
    """
    新增數(shù)據(jù)
    """
 
    try:
      self.cursor_interface_testing.execute(sql,data)
      self.connect_interface_testing.commit()
      res = {data,'新增成功'}
    except Exception as e:
      res = {'新增失敗',e}
    return res
  def conn_close(self):
    # 關(guān)閉數(shù)據(jù)庫(kù)
    self.cursor_interface_testing.close()
 
 
if __name__ == "__main__":
  # ()類的實(shí)例化
  om = OperateMysql()
 
  # 新增
  data = [{'id': 1, 'name': '測(cè)試', 'age': 15}, {'id': 2, 'name': '老王', 'age': 10}, {'id': 3, 'name': '李四', 'age': 20}]
  for i in data:
    i_data = (i['id'],i['name'],i['age'])
    insert_res = om.insert_data(
      """
       INSERT INTO test_student (id,name,age) VALUES (%s,%s,%s)
      """,i_data
    )
    print(insert_res)
 
  # 查詢
  one_data = om.select_first_data(
    """
      SELECT * FROM test_student;
    """
  )
  all_data = om.select_all_data(
    """
    SELECT * FROM test_student;
    """
  )
  print(one_data)
  # all_data字符串類型的list轉(zhuǎn)list
  print("查詢總數(shù)據(jù):%s",len(json.loads(all_data)),"分別是:%s",all_data)
 
  # 修改
  update_data = om.update_data(
    """
    UPDATE test_student SET name = '王五' WHERE id = 1;
    """
  )
  print(update_data)
 
  # 刪除
  del_data = om.del_data(
    """
    DELETE FROM test_student WHERE id in (1,2,3);
    """
  )
  print(del_data)
 
  # 關(guān)閉游標(biāo)
  om.conn_close()