base_db_service.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@Email: ludq1@chinaunicom.cn
@date: 2023/04/07 11:40:00
@Description:
"""

import logging
from typing import Optional

from pymysql.connections import Connection
from pymysql.cursors import DictCursor

from .app_exception import AppRuntimeException
from .common_app_config import CommonAppConfig
from ..globalutility import Utility
from ..my_stringutils import MyStringUtils


class QueryResultWithFoundRows:
    result_list: Optional[list] = None
    found_rows: Optional[int] = None

    def __init__(self, **kwargs):
        self.result_list = kwargs.get("result_list")
        self.found_rows = kwargs.get("found_rows")

    def __str__(self):
        return Utility.dict2jsonstr(self.__dict__)


class DbBehaviourAfterExec:
    r"""
    数据库语句执行完成后的连接操作行为，不依赖任何其他应用基础类
    默认 commit_after_exec 和 close_conn_after_exec 为 False
    close_conn_when_exception 和 rollback_when_exception 为 True ，commit_when_exception 为 False
    rollback_when_exception 的优先级比 commit_when_exception 高，
    当 close_conn_when_exception 为True时，如果 rollback_when_exception 和 commit_when_exception 都为False，
    则执行 rollback 操作
    """
    commit_after_exec: bool = True
    close_conn_after_exec: bool = True
    close_conn_when_exception: bool = True
    rollback_when_exception: bool = True
    commit_when_exception: bool = False

    def __init__(self, commit_after_exec: bool = True, close_conn_after_exec: bool = True,
                 close_conn_when_exception: bool = True, rollback_when_exception: bool = True,
                 commit_when_exception: bool = False):
        r"""
        初始化数据库行为

        :param commit_after_exec:
        :param close_conn_after_exec:
        :param close_conn_when_exception:
        :param rollback_when_exception:
        :param commit_when_exception:
        """
        self.commit_after_exec = commit_after_exec
        self.close_conn_after_exec = close_conn_after_exec
        self.close_conn_when_exception = close_conn_when_exception
        self.rollback_when_exception = rollback_when_exception
        self.commit_when_exception = commit_when_exception

    @classmethod
    def gen_no_commit_close_behaviour(cls):
        r"""
        生成一个不commit，不close，异常时rollback、close的数据库行为

        :return:
        """
        return DbBehaviourAfterExec(
            commit_after_exec=False,
            close_conn_after_exec=False,
            close_conn_when_exception=True,
            rollback_when_exception=True,
            commit_when_exception=False)


class BaseDbService:
    r"""
    基础 dbService类,定义一些基础函数，不依赖任何其他应用基础类
    """

    def __init__(self, logger: logging.Logger = None):
        self.logger = logger or CommonAppConfig().common_logger

    def join_str(self, *to_join_list: str, separator_str: str = '', wrapper_str: str = None,
                 ignore_none: bool = False) -> str:
        r"""
        连接 list， 当值为 None 时当做 null字符串 添加，但是不追加 wrapper_str

        :param to_join_list
        :param separator_str 默认使用空字符串
        :param wrapper_str
        :param ignore_none 默认为True
        :return
        """
        return Utility.list_join_to_str(to_join_list, separator_str=separator_str, wrapper_str=wrapper_str,
                                        ignore_none=ignore_none)

    def do_filter_mysql_param_with_single_quotes(self, to_filter_param: str, append_fix: str = "'") -> str:
        r"""
        mysql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对value中的特殊字符进行替换

        :param to_filter_param
        :param append_fix 默认为单引号
        :return
        """
        return Utility.do_filter_mysql_param(to_filter_param=to_filter_param, append_fix=append_fix)

    def do_filter_mysql_param_with_single_quotes_for_fuzzy_search(self, to_filter_param: str,
                                                                  append_fix: str = "'") -> str:
        r"""
        mysql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对value中的特殊字符进行替换

        :param to_filter_param
        :param append_fix 默认为单引号
        :return
        """
        return Utility.do_filter_mysql_param_with_single_quotes_for_fuzzy_search(to_filter_param=to_filter_param,
                                                                                 append_fix=append_fix)

    def do_filter_mysql_param_list_with_single_quotes(
            self,
            to_filter_param_list: list,
            append_fix: str = "'"
    ) -> Optional[list]:
        r"""
        mysql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对value中的特殊字符进行替换

        Utility.do_filter_mysql_param_list_with_single_quotes函数有bug，所以自己实现一下

        :param to_filter_param_list
        :param append_fix 默认为单引号
        :return
        """
        if to_filter_param_list is None:
            return None

        return [self.do_filter_mysql_param_with_single_quotes(tmp_key, append_fix=append_fix) for tmp_key in
                to_filter_param_list]

    def gen_default_db_behaviour_after_exec(self):
        r"""
        生成默认的 db_behaviour_after_exec

        默认行为为 正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :return:
        """
        return DbBehaviourAfterExec(
            commit_after_exec=True, close_conn_after_exec=True,
            close_conn_when_exception=True,
            rollback_when_exception=True,
            commit_when_exception=False)

    def commit(self, conn: Connection, close_conn_after_exec: bool = True,
               db_behaviour_after_exec: DbBehaviourAfterExec = None) -> bool:
        r"""
        执行commit操作

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param close_conn_after_exec
        :param db_behaviour_after_exec
        :return
        """
        if conn is None:
            return True

        conn.commit()
        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        if close_conn_after_exec or (db_behaviour_after_exec and db_behaviour_after_exec.close_conn_after_exec):
            conn.close()

        return True

    def rollback(self, conn: Connection, close_conn_after_exec: bool = True,
                 db_behaviour_after_exec: DbBehaviourAfterExec = None) -> bool:
        r"""
        执行rollback操作

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param close_conn_after_exec
        :param db_behaviour_after_exec
        :return
        """
        if conn is None:
            return True

        conn.rollback()
        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        if close_conn_after_exec or (db_behaviour_after_exec and db_behaviour_after_exec.close_conn_after_exec):
            conn.close()

        return True

    def close_conn(self, conn: Connection):
        r"""
        关闭连接

        :param conn
        """
        if conn is None:
            return

        conn.close()

    def do_after_exec(self, conn: Connection, result, encountered_exception,
                      db_behaviour_after_exec: DbBehaviourAfterExec):
        r"""
        数据库执行完毕后的行为封装

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param result
        :param encountered_exception
        :param db_behaviour_after_exec
        :return:
        """

        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        if not encountered_exception:
            return self.do_after_exec_without_exception(conn, result, db_behaviour_after_exec)
        else:
            return self.do_after_exec_with_exception(conn, encountered_exception, db_behaviour_after_exec)

    def do_after_exec_without_exception(self, conn: Connection, result, db_behaviour_after_exec: DbBehaviourAfterExec):
        r"""
        数据库执行完毕后的行为封装

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param result
        :param db_behaviour_after_exec
        :return:
        """
        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        close_conn_after_exec = True if db_behaviour_after_exec.close_conn_after_exec else False
        commit_after_exec = True if db_behaviour_after_exec and db_behaviour_after_exec.commit_after_exec else False
        if commit_after_exec and conn is not None:
            conn.commit()

        if close_conn_after_exec and conn is not None:
            conn.close()

        return result

    def do_after_exec_with_exception(self, conn: Connection, encountered_exception,
                                     db_behaviour_after_exec: DbBehaviourAfterExec):
        r"""
        数据库执行完毕后的行为封装

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param encountered_exception
        :param db_behaviour_after_exec
        :return:
        """
        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        close_conn_when_exception = True if db_behaviour_after_exec.close_conn_when_exception else False
        rollback_when_exception = True if db_behaviour_after_exec.rollback_when_exception else False
        commit_when_exception = True if db_behaviour_after_exec.commit_when_exception else False
        if close_conn_when_exception and not rollback_when_exception and not commit_when_exception:
            # 当异常时关闭连接 但是 rollback_when_exception 和 commit_when_exception 都为False，则默认使用 rollback
            rollback_when_exception = True

        if rollback_when_exception and conn is not None:
            conn.rollback()
        elif commit_when_exception and conn is not None:
            conn.commit()

        if close_conn_when_exception and conn is not None:
            conn.close()

        raise encountered_exception

    def querydb(
            self, conn: Connection, query_str: str, factory_method=None,
            db_behaviour_after_exec: DbBehaviourAfterExec = None
    ) -> list:
        r"""
        查询数据库

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param query_str
        :param factory_method
        :param db_behaviour_after_exec
        :return
        """

        if conn is None:
            raise AppRuntimeException(message="未提供conn", detail="执行数据库操作时，传入的conn为None")

        encountered_exception: Optional[BaseException] = None
        result: Optional[list] = None
        try:
            self.logger.debug(f"执行sql语句:{query_str}")
            with conn.cursor(cursor=DictCursor) as cursor:
                try:
                    cursor.execute(query_str)
                    result = cursor.fetchall() or list()
                finally:
                    cursor.close()
            self.logger.debug(f"执行sql语句:{query_str} 返回结果: {result}")
        except BaseException as e:
            encountered_exception = e
            self.logger.debug(f"执行sql语句:{query_str} 发生异常: {e}")

        if not encountered_exception and factory_method is not None and result:
            try:
                result = [factory_method(tmpDict) for tmpDict in result]
            except BaseException as e:
                encountered_exception = e

        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        return self.do_after_exec(conn, result, encountered_exception, db_behaviour_after_exec)

    def querycount_db(self, conn: Connection, query_str: str,
                      db_behaviour_after_exec: DbBehaviourAfterExec = None) -> int:
        r"""
        查询数据库,执行count数据库查询,直接得到返回的整数

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param query_str
        :param db_behaviour_after_exec
        :return
        """

        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        ret_list = self.querydb(conn, query_str, db_behaviour_after_exec=db_behaviour_after_exec)

        if not ret_list:
            raise ValueError('数据库查询语句返回结果集为空')

        return int(list(ret_list[0].values())[0])

    def querydb_with_single_column_and_line(self, conn: Connection, query_str: str,
                                            db_behaviour_after_exec: DbBehaviourAfterExec = None) -> Optional[str]:
        r"""
        查询数据库,执行返回单列单行的数据，并获取字符串,查询的数据不存在时，返回None

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param query_str
        :param db_behaviour_after_exec
        :return
        """

        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        ret_list = self.querydb(conn, query_str, db_behaviour_after_exec=db_behaviour_after_exec)

        if not ret_list:
            return None

        return MyStringUtils.to_str(list(ret_list[0].values())[0])

    def querydb_with_sql_calc_found_rows(
            self, conn: Connection, query_str: str, factory_method=None,
            db_behaviour_after_exec: DbBehaviourAfterExec = None
    ) -> QueryResultWithFoundRows:
        r"""
        查询数据库,为带有 SQL_CALC_FOUND_ROWS 的sql语句返回 FOUND_ROWS()

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param query_str
        :param factory_method
        :param db_behaviour_after_exec
        :return
        """
        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        tmp_db_behaviour_after_exec = DbBehaviourAfterExec.gen_no_commit_close_behaviour()
        tmp_db_behaviour_after_exec.close_conn_when_exception = db_behaviour_after_exec.close_conn_when_exception
        tmp_db_behaviour_after_exec.commit_when_exception = db_behaviour_after_exec.commit_when_exception
        tmp_db_behaviour_after_exec.rollback_when_exception = db_behaviour_after_exec.rollback_when_exception

        ret_list = self.querydb(conn, query_str, factory_method=factory_method,
                                db_behaviour_after_exec=tmp_db_behaviour_after_exec)

        found_rows = self.querydb_with_single_column_and_line(
            conn=conn,
            query_str="select FOUND_ROWS() as found_rows",
            db_behaviour_after_exec=db_behaviour_after_exec
        )
        found_rows = int(found_rows)
        return QueryResultWithFoundRows(result_list=ret_list, found_rows=found_rows)

    def operate_db(self, conn: Connection, *operate_sql_list: str, sql_list: list = None,
                   db_behaviour_after_exec: DbBehaviourAfterExec = None) -> int:
        r"""
        操作数据库,返回受影响的行数, 可以有两种提供操作语句的方式,或者直接执行多条语句，或者使用 sql_list 提供操作语句列表

        当 db_behaviour_after_exec为None时，使用默认的 db_behaviour_after_exec，即正常操作后commit、关闭数据库， 异常时rollback、关闭数据库

        :param conn
        :param operate_sql_list
        :param sql_list
        :param db_behaviour_after_exec
        :return
        """
        if conn is None:
            raise AppRuntimeException(message="未提供conn", detail="执行数据库操作时，传入的conn为None")

        encountered_exception: Optional[BaseException] = None
        result: int = 0
        try:
            # 先生成列表
            real_sql_list = []
            if operate_sql_list:
                real_sql_list += operate_sql_list
            elif sql_list:
                real_sql_list += sql_list
            else:
                raise ValueError('未提供sql语句')

            with conn.cursor(cursor=DictCursor) as cursor:
                try:
                    for tmp_sql_str in real_sql_list:
                        self.logger.debug(f"执行sql语句:{tmp_sql_str}")
                        result += cursor.execute(tmp_sql_str)
                        self.logger.debug(f"执行sql语句:{tmp_sql_str} 返回结果: {result}")
                finally:
                    cursor.close()
        except BaseException as e:
            encountered_exception = e

        if db_behaviour_after_exec is None:
            db_behaviour_after_exec = self.gen_default_db_behaviour_after_exec()

        return self.do_after_exec(conn, result, encountered_exception, db_behaviour_after_exec)