Cator

pip install cator

import cator

# mysql
db_url = "mysql://root:123456@127.0.0.1:3306/data?charset=utf8&autocommit=true"

# sqlite
db_url = 'sqlite:///data.db?isolation_level=null'

# open Database
db = cator.connect(db_url)

# close
db.close()

class Connection(ABC):
    def close(self):
        pass

    def commit(self):
        pass

    def rollback(self):
        pass

    def cursor(self):
        pass

from peewee import MySQLDatabase
from cator import DatabaseProxy


config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': '123456',
    'database': 'data',
    'charset': 'utf8mb4',
}

db = MySQLDatabase(**config)

# use cator database proxy
db_proxy = DatabaseProxy(db)

import pymysql
from cator import DatabaseProxy

config = {
    'host': '127.0.0.1',
    'user': 'root',
    'password': '123456',
    'database': 'data',
    'port': 3306
}

connection = pymysql.connect(**config)
proxy_db = DatabaseProxy(connection)

rows = proxy_db.select('select * from person where id = :id', {'id': 15})
print(rows)
proxy_db.close()

class DatabaseProxy:
    def table(self, table_name):
        pass
    
    def select(self, operation, params=()):
        pass

    def select_one(self, operation, params=()):
        pass

    def update(self, operation, params=()):
        pass

    def delete(self, operation, params=()):
        pass

    def insert(self, operation, params: Union[list, dict]):
        pass

    def insert_one(self, operation, params: Union[tuple, dict] = ()):
        pass

    def before_execute(self, operation, params=None):
        pass

    def after_execute(self, cursor):
        pass

    def execute(self, operation, params=None):
        pass
    
    def cursor(self, *args, **kwargs):
        """return cursor object"""

    def connect(self):
        """connect database"""

    def close(self):
        """close connection"""
        
    def commit(self):
        pass

    def rollback(self):
        pass

class Table:

    @property
    def total(self):
        pass
        
    def insert(self, data: Union[dict, list]):
        pass
        
    def insert_one(self, data: dict):
        pass
        
    def delete_by_id(self, uid):
        pass
        
    def update_by_id(self, uid, data):
        pass
        
    def select_by_id(self, uid):
        pass