# -*-coding:utf-8 -*- # # --------------------------------------------------------------------------- # ProjectName: test62 # FileName: db.py # Author: lao_zhao # Datetime: 2024/9/4 14:35 # Description: # # --------------------------------------------------------------------------- import pymysql from apiAutoTest_v3 import log from apiAutoTest_v3.common.basic_read_ini import ReadIni class DB: def __init__(self): """链接数据库,获取链接对象和游标对象""" ini = ReadIni() self.conn = pymysql.connect( host=ini.get_sql_connect_msg("host"), port=int(ini.get_sql_connect_msg("port")), user=ini.get_sql_connect_msg("user"), password=ini.get_sql_connect_msg("password"), database=ini.get_sql_connect_msg("database"), charset="utf8" ) self.cursor = self.conn.cursor() def close(self): self.cursor.close() self.conn.close() def select(self, sql, n=0): """执行查询的sql语句,并返回查询的结果""" if isinstance(sql, str): if sql.strip().lower().startswith("select"): try: self.cursor.execute(sql) except Exception as e: log.error(f"执行select方法时,产生错误,错误为:{e},sql语句为:{sql}") raise e else: select_result = self.cursor.fetchall() if isinstance(n, int) and select_result: try: return select_result[n][0] except Exception as e: log.error(f"执行select方法时,获取查询结果失败,传入的n为:{n}") raise e else: log.error(f"执行select方法时,sql语句不是查询的sql语句,sql语句为:{sql}") raise ValueError("sql语句错误") else: log.error(f"执行select方法时,sql语句错误,sql语句为:{sql}") raise ValueError("sql语句错误") def delete(self, sql): """执行删除的SQL语句""" if isinstance(sql, str): if sql.strip().lower().startswith("delete"): try: self.cursor.execute(sql) self.conn.commit() except Exception as e: log.error(f"执行delete方法时,产生错误,sql语句为:{sql}, 错误为:{e}") raise e else: log.error(f"执行delete方法时,sql语句错误,sql语句为:{sql}") raise ValueError("sql语句错误") else: log.error(f"执行delete方法时,sql语句错误,sql语句为:{sql}") raise ValueError("sql语句错误")