db.py

# -*-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.read_project_ini import ReadProjectIni


class DB:
    def __init__(self):
        """链接数据库，获取链接对象和游标对象"""
        # 链接数据库，数据库的连续配置信息存放在项目的配置文件中的，需要使用ReadProjectIni类对象调用get_sql_connect_msg方法
        pro_ini = ReadProjectIni()
        self.conn = pymysql.connect(
            host=pro_ini.get_sql_connect_msg("host"),
            port=int(pro_ini.get_sql_connect_msg("port")),
            user=pro_ini.get_sql_connect_msg("user"),
            password=pro_ini.get_sql_connect_msg("password"),
            database=pro_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语句错误")