Mysql.py

import MySQLdb
from MySQLdb.cursors import DictCursor
from dbutils.pooled_db import PooledDB
from config.db_config import DbConfig


class Mysql(object):
    # Connection pool object
    __pool = None

    def __init__(self, db: DbConfig):
        # The database constructor takes the connection out of the connection pool and generates the operation cursor
        self._conn = Mysql.__get_conn(db)
        self._cursor = self._conn.cursor()

    @staticmethod
    def __get_conn(db: DbConfig):
        """
        @summary: Static method to remove connections from connection pool
        @return MySQLdb.connection
        """
        if Mysql.__pool is None:
            __pool = PooledDB(creator=MySQLdb, mincached=1, maxcached=20,
                              host=db.host, port=int(db.port), user=db.username, passwd=db.password,
                              db=db.db_name, use_unicode=False, cursorclass=DictCursor,
                              setsession=['SET AUTOCOMMIT = 1'])
        return __pool.connection()

    def get_all(self, sql, param=None):
        """
        @summary: Execute the query and fetch all result sets
        @param sql:queryＳＱＬ，If there are query conditions, specify only the list of conditions and use parameters for the condition values[param]Pass in
        @param param: Optional parameters, conditional list values (tuples)/List)
        @return: result list(Dictionary object)/boolean Queried result sets
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchall()
        else:
            result = False
        return result

    def get_one(self, sql, param=None):
        """
        @summary: Execute the query and take out Article 1
        @param sql:queryＳＱＬ，If there are query conditions, specify only the list of conditions and use parameters for the condition values[param]Pass in
        @param param: Optional parameters, conditional list values (tuples)/List)
        @return: result list/boolean Queried result sets
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchone()
        else:
            result = False
        return result

    def get_many(self, sql, num, param=None):
        """
        @summary: Execute the query and take it out num Article result
        @param sql:queryＳＱＬ，If there are query conditions, specify only the list of conditions and use parameters for the condition
        @param num:Number of results obtained
        @param param: Optional parameters, conditional list values (tuples)/List)
        @return: result list/boolean Queried result sets
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchmany(num)
        else:
            result = False
        return result

    def insert_one(self, sql, value):
        """
        @summary: Insert a record into the data table
        @param sql:To insertＳＱＬformat
        @param value:Record data to be inserted tuple/list
        @return: insertId Number of rows affected
        """
        self._cursor.execute(sql, value)
        return self.__getInsertId()

    def insert_many(self, sql, values):
        """
        @summary: Insert multiple records into the data table
        @param sql:To insertＳＱＬformat
        @param values:Record data to be inserted tuple(tuple)/list[list]
        @return: count Number of rows affected
        """
        count = self._cursor.executemany(sql, values)
        return count

    def __get_insert_id(self):
        """
        Gets the last insert operation generated by the current connection id,If not, 0
        """
        self._cursor.execute("SELECT @@IDENTITY AS id")
        result = self._cursor.fetchall()
        return result[0]['id']

    def __query(self, sql, param=None):
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        return count

    def update(self, sql, param=None):
        """
        @summary: Update data table records
        @param sql: ＳＱＬFormat and conditions, use(%s,%s)
        @param param: To be updated  value tuple/list
        @return: count Number of rows affected
        """
        return self.__query(sql, param)

    def delete(self, sql, param=None):
        """
        @summary: Delete data table records
        @param sql: ＳＱＬFormat and conditions, use(%s,%s)
        @param param: Conditions to be deleted value tuple/list
        @return: count Number of rows affected
        """
        return self.__query(sql, param)

    def begin(self):
        """
        @summary: Open a transaction
        """
        self._conn.autocommit(1)

    def end(self, option='commit'):
        """
        @summary: Closing the transaction
        """
        if option == 'commit':
            self._conn.commit()
        else:
            self._conn.rollback()

    def dispose(self, is_end=1):
        """
        @summary: Release connection pool resources
        """
        if is_end == 1:
            self.end('commit')
        else:
            self.end('rollback')
        self._cursor.close()
        self._conn.close()