db_connector_basic.py

"""An empty database object on which to implement a database I/O module."""

from abc import ABCMeta
from abc import abstractmethod


class Database(metaclass=ABCMeta):
    """Methods and properties required for database connection are defined."""

    def __enter__(self):
        """This function is called when the'with' statement is started."""
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        """This function is called when the'with' statement ends."""
        pass

    def connect(self, host, port, user, password):
        """Connect to the database.  
        This is mainly called by the`__enter__` function.  
        Below are the recommended basic arguments.

        Args:
            host (str): The address of the database.
            port (int): The port number of the database.
            user (str): Username of the database.
            password (str): Password of the database.
        """
        pass

    def disconnect(self):
        """Disconnect to the database.  
        This is mainly called by the`__exit__` function.
        """
        pass

    @abstractmethod
    def execute(self, query, args):
        """Execute one SQL query  
        Below are the recommended basic arguments.

        Args:
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

    @abstractmethod
    def executemany(self, query, args):
        """Execute multiple SQL queries.  
        Below are the recommended basic arguments.

        Args:
            query (str): The SQL query.
            args (any): These are the arguments to be filled in the SQL query.
        """
        pass

    @abstractmethod
    def execute_and_fetchone(self, query, args):
        """Execute one SQL query and get one result.  
        Below are the recommended basic arguments.

        Args:
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

    @abstractmethod
    def execute_and_fetchmany(self, query, args):
        """Execute one SQL query and get multiple results.  
        Below are the recommended basic arguments.

        Args:
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

    @abstractmethod
    def execute_and_fetchall(self, query, args):
        """Execute one SQL query and get all results.  
        Below are the recommended basic arguments.

        Args:
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

    def execute_save_json(self, path, query, args):
        """Execute one SQL query and save result as JSON.  
        Below are the recommended basic arguments.

        Args:
            path (str): Path where the file will be saved
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

    def execute_save_yaml(self, query, args):
        """Execute one SQL query and save result as YAML.  
        Below are the recommended basic arguments.

        Args:
            path (str): Path where the file will be saved
            query (str): The SQL query.
            args (any): It is used when the SQL query is written in statement format.
        """
        pass

mysql_connector.py

import datetime
import json
import pymysql
import yaml
from db_connector_basic import Database

class MySQL(Database):
    """MySQL Connector"""
    def __init__(self, host='localhost', port=3306, user=None, password=None, db=None, charset=None):
        self.__host = host
        self.__port = port
        self.__user = user
        self.__password = password
        self.__db = db
        self.__charset = charset
        self.__connection = pymysql.connect(
            host=self.__host,
            port=self.__port,
            user=self.__user,
            password=self.__password,
            db=self.__db,
            charset=self.__charset,
        )
        self.__cursor = self.__connection.cursor(pymysql.cursors.DictCursor)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def __datetime_to_str(self, query_res):
        if isinstance(query_res, dict):
            query_res = [query_res]

        for i, res in enumerate(query_res):
            for key, value in res.items():
                if isinstance(value, datetime.datetime):
                    query_res[i][key] = value.strftime(r'%Y-%m-%d %H:%M:%S')
        return query_res

    # Getter and Setters
    @property
    def connection(self):
        return self.__connection

    @property
    def cursor(self):
        return self.__cursor

    @property
    def host(self):
        return self.__host
    
    @property
    def port(self):
        return self.__port

    @property
    def user(self):
        return self.__user

    @property
    def password(self):
        return self.__password

    @property
    def db(self):
        return self.__db

    @property
    def charset(self):
        return self.__charset

    @host.setter
    def host(self, host):
        self.__host = host

    @port.setter
    def port(self, port):
        self.__port = port

    @user.setter
    def user(self, user):
        self.__user = user

    @password.setter
    def password(self, password):
        self.__password = password

    @db.setter
    def db(self, db):
        self.__db = db

    @charset.setter
    def charset(self, charset):
        self.__charset = charset

    def commit(self):
        self.connection.commit()

    def close(self, commit=True):
        if commit:
            self.commit()
        self.connection.close()

    def execute(self, query: str, *, args={}):
        try:
            self.cursor.execute(query=query, args=args)
        except pymysql.MySQLError as e:
            print(e)

    def executemany(self, query: str, *, args=[]):
        self.cursor.executemany(query=query, args=args)

    def execute_and_fetchone(self, query: str, *, args={}):
        self.execute(query=query, args=args)
        return self.__datetime_to_str(self.cursor.fetchone())
    
    def execute_and_fetchmany(self, query: str, *, args={}, size=1):
        self.execute(query=query, args=args)
        return self.__datetime_to_str(self.cursor.fetchmany(size=size))

    def execute_and_fetchall(self, query: str, *, args={}):
        self.execute(query=query, args=args)
        return self.__datetime_to_str(self.cursor.fetchall())

    def __execute_and_fetch(self, query: str, *, args={}, size):
        res = []
        if size <= 0:
            res = self.execute_and_fetchall(query=query, args=args)
        elif size == 1:
            res = self.execute_and_fetchone(query=query, args=args)
        else:
            res = self.execute_and_fetchmany(query=query, args=args, size=size)
        return res

    def execute_save_json(self, path: str, query: str, *, args={}, size=0):
        res = self.__execute_and_fetch(query=query, args=args, size=size)
        with open(file=path, mode='w', encoding='utf-8') as f:
            f.write(json.dumps(res, indent='\t', ensure_ascii=False))
        return path

    def execute_save_yaml(self, path: str, query: str, *, args={}, size=0):
        res = self.__execute_and_fetch(query=query, args=args, size=size)
        with open(file=path, mode='w', encoding='utf-8') as f:
            f.write(yaml.dump(res, indent=4, allow_unicode=True))
        return path

if __name__ == "__main__":
    with MySQL(host=HOSTNAME, port=PORTNUMBER, user=USERNAME, password=PASSWORD, db=DBNAME, charset='utf8mb4') as mysql:
        res = mysql.execute_and_fetchall('SELECT * FROM users')
    print(res)