qc-dgp/dgp/tests/MySQLConnection.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
__title__ = 'MySQL连接池对象'

@Time    : 2020/9/26 18:36
@Author  : Kenny-PC
@Software: PyCharm

# code is far away from bugs with the god animal protecting
    I love animals. They taste delicious.
              ┏┓      ┏┓
            ┏┛┻━━━┛┻┓
            ┃      ☃      ┃
            ┃  ┳┛  ┗┳  ┃
            ┃      ┻      ┃
            ┗━┓      ┏━┛
                ┃      ┗━━━┓
                ┃  神兽保佑    ┣┓
                ┃　永无BUG！   ┏┛
                ┗┓┓┏━┳┓┏┛
                  ┃┫┫  ┃┫┫
                  ┗┻┛  ┗┻┛
"""
"""
数据库连接池相关
"""

import pymysql
from DBUtils.PooledDB import PooledDB
import logging
import configparser
import os

# 获取当前文件所在目录的上一级目录
root_dir = os.path.dirname(os.path.abspath('.'))
print('当前项目根目录为：', root_dir)

# 读取数据库配置信息
config=configparser.ConfigParser()
config.read(root_dir + '/tests/conf/db.ini',encoding='UTF-8')
sections=config.sections()

# 数据库工厂
dbFactory={}
for dbName in sections:
    # 读取相关属性
    maxconnections=config.get(dbName,"maxconnections")
    mincached=config.get(dbName,"mincached")
    maxcached=config.get(dbName,"maxcached")
    host=config.get(dbName,"host")
    port=config.get(dbName,"port")
    user=config.get(dbName,"user")
    password=config.get(dbName,"password")
    database=config.get(dbName,"database")
    databasePooled=PooledDB(creator=pymysql,
                        maxconnections=int(maxconnections),
                        mincached=int(mincached),
                        maxcached=int(maxcached),
                        blocking=True,
                        cursorclass = pymysql.cursors.DictCursor,
                        host=host,
                        port=int(port),
                        user=user,
                        password=password,
                        database=database)
    dbFactory[dbName]=databasePooled

class MySQLConnection(object):
    """
    数据库连接池代理对象
    查询参数主要有两种类型
    第一种：传入元祖类型,例如(12,13),这种方式主要是替代SQL语句中的%s展位符号
    第二种: 传入字典类型,例如{"id":13},此时我们的SQL语句需要使用键来代替展位符,例如：%(name)s
    """
    def __init__(self,dbName="test"):
        self.connect = dbFactory[dbName].connection()
        self.cursor = self.connect.cursor()
        logging.debug("获取数据库连接对象成功,连接池对象:{}".format(str(self.connect)))

    def execute(self,sql,param=None):
        """
        基础更新、插入、删除操作
        :param sql:
        :param param:
        :return: 受影响的行数
        """
        ret=None
        try:
            if param==None:
                ret=self.cursor.execute(sql)
            else:
                ret=self.cursor.execute(sql,param)
        except TypeError as te:
            logging.debug("类型错误")
            logging.exception(te)
        return ret
    def query(self,sql,param=None):
        """
        查询数据库
        :param sql: 查询SQL语句
        :param param: 参数
        :return: 返回集合
        """
        self.cursor.execute(sql,param)
        result=self.cursor.fetchall()
        return result
    def queryOne(self,sql,param=None):
        """
        查询数据返回第一条
        :param sql: 查询SQL语句
        :param param: 参数
        :return: 返回第一条数据的字典
        """
        result=self.query(sql,param)
        if result:
            return result[0]
        else:
            return None
    def listByPage(self,sql,current_page,page_size,param=None):
        """
        分页查询当前表格数据
        :param sql: 查询SQL语句
        :param current_page: 当前页码
        :param page_size: 页码大小
        :param param:参数
        :return:
        """
        countSQL="select count(*) ct from ("+sql+") tmp "
        logging.debug("统计SQL:{}".format(sql))
        countNum=self.count(countSQL,param)
        offset=(current_page-1)*page_size
        totalPage=int(countNum/page_size)
        if countNum % page_size>0:
            totalPage = totalPage + 1
        pagination={"current_page":current_page,"page_size":page_size,"count":countNum,"total_page":totalPage}
        querySql="select * from ("+sql+") tmp limit %s,%s"
        logging.debug("查询SQL:{}".format(querySql))
        # 判断是否有参数
        if param==None:
            # 无参数
            pagination["data"]=self.query(querySql,(offset,page_size))
        else:
            # 有参数的情况,此时需要判断参数是元祖还是字典
            if isinstance(param,dict):
                # 字典的情况,因此需要添加字典
                querySql="select * from ("+sql+") tmp limit %(tmp_offset)s,%(tmp_pageSize)s"
                param["tmp_offset"]=offset
                param["tmp_pageSize"]=page_size
                pagination["data"]=self.query(querySql,param)
            elif isinstance(param,tuple):
                # 元祖的方式
                listtp=list(param)
                listtp.append(offset)
                listtp.append(page_size)
                pagination["data"]=self.query(querySql,tuple(listtp))
            else:
                # 基础类型
                listtp=[]
                listtp.append(param)
                listtp.append(offset)
                listtp.append(page_size)
                pagination["data"]=self.query(querySql,tuple(listtp))
        return pagination
    def count(self,sql,param=None):
        """
        统计当前表记录行数
        :param sql: 统计SQL语句
        :param param: 参数
        :return: 当前记录行
        """
        ret=self.queryOne(sql,param)
        count=None
        if ret:
            for k,v in ret.items():
                count=v
        return count

    def insert(self,sql,param=None):
        """
        数据库插入
        :param sql: SQL语句
        :param param: 参数
        :return: 受影响的行数
        """
        return self.execute(sql,param)
    def update(self,sql,param=None):
        """
        更新操作
        :param sql: SQL语句
        :param param: 参数
        :return: 受影响的行数
        """
        return self.execute(sql,param)
    def delete(self,sql,param=None):
        """
        删除操作
        :param sql: 删除SQL语句
        :param param: 参数
        :return: 受影响的行数
        """
        return self.execute(sql,param)
    def batch(self,sql,param=None):
        """
        批量插入
        :param sql: 插入SQL语句
        :param param: 参数
        :return: 受影响的行数
        """
        return self.cursor.executemany(sql,param)
    def commit(self,param=None):
        """
        提交数据库
        :param param:
        :return:
        """
        if param==None:
            self.connect.commit()
        else:
            self.connect.rollback()

    def close(self):
        """
        关闭数据库连接
        :return:
        """
        if self.cursor:
            self.cursor.close()
        if self.connect:
            self.connect.close()
        logging.debug("释放数据库连接")
        return None