db.py 3.03 KB
Newer Older
crob committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74
# -*-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语句错误")