# -*-coding:utf-8 -*- # # --------------------------------------------------------------------------- # ProjectName: test63 # FileName: db.py # Author: lao_zhao # Datetime: 2024/11/5 14:22 # Description: # # --------------------------------------------------------------------------- import pymysql from APIAutoTest_v3_1_2 import log from APIAutoTest_v3_1_2.common.read_basic_ini import ReadBasicIni class DB: def __init__(self): """链接数据库""" ini = ReadBasicIni() try: self.conn = pymysql.connect( host=ini.get_connect_sql_msg("host"), port=int(ini.get_connect_sql_msg("port")), user=ini.get_connect_sql_msg("user"), password=ini.get_connect_sql_msg("pwd"), database=ini.get_connect_sql_msg("database"), charset="utf8" ) self.cursor = self.conn.cursor() except Exception as e: log.error(f"链接数据库时出错,请求检查ini配置文件,发生的错误类型为:{type(e)}, 错误的描述为:{e}") raise e def delete(self, sql_sentence): """执行删除的sql语句""" log.info(f"执行的方法为:delete,功能描述为:执行删除的sql语句,传入的参数为:{sql_sentence}") if isinstance(sql_sentence, str) and sql_sentence.strip().lower().startswith("delete"): try: self.cursor.execute(sql_sentence) self.conn.commit() except Exception as e: log.error( f"执行的方法为:delete,功能描述为:执行删除的sql语句,传入的参数为:{sql_sentence}, 错误类型为:{type(e)}, 错误的描述为:{e}") raise e else: log.error(f"执行的方法为:delete,功能描述为:执行删除的sql语句,传入的参数为:{sql_sentence}, 错误类型为:ValueError, 错误的描述为:sql语句错误") raise ValueError("sql语句错误") def select(self, sql_sentence): """执行查询的sql语句""" log.info(f"执行的方法为:select,功能描述为:执行查询的sql语句,传入的参数为:{sql_sentence}") if isinstance(sql_sentence, str) and sql_sentence.strip().lower().startswith("select"): try: self.cursor.execute(sql_sentence) except Exception as e: log.error( f"执行的方法为:select,功能描述为:执行查询的sql语句,传入的参数为:{sql_sentence}, 错误类型为:{type(e)}, 错误的描述为:{e}") raise e else: # 接收查询的结果 select_result = self.cursor.fetchall() # 判断查询结果是否为空的元组,如果为空的元组返回None,如果不为空的元组返回查询的第一个结果 if select_result: return select_result[0][0] else: log.error(f"执行的方法为:select,功能描述为:执行查询的sql语句,传入的参数为:{sql_sentence}, 错误类型为:ValueError, 错误的描述为:sql语句错误") raise ValueError("sql语句错误")