60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
|
import typing
|
||
|
|
|
||
|
|
import pymysql
|
||
|
|
|
||
|
|
|
||
|
|
class Database:
|
||
|
|
__instance: 'Database' = None
|
||
|
|
__client: pymysql.Connection = None
|
||
|
|
|
||
|
|
def __new__(cls, *args, **kwargs):
|
||
|
|
if cls.__instance is None:
|
||
|
|
cls.__instance = super().__new__(cls)
|
||
|
|
return cls.__instance
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.__create_con()
|
||
|
|
|
||
|
|
def __create_con(self):
|
||
|
|
# TODO Set to environment variables. Use pydantic-settings
|
||
|
|
self.__client = pymysql.connect(
|
||
|
|
host='127.0.0.1',
|
||
|
|
port=3306,
|
||
|
|
user='root',
|
||
|
|
password='root',
|
||
|
|
db='jde',
|
||
|
|
charset='utf8'
|
||
|
|
)
|
||
|
|
|
||
|
|
def __execute_cmd(self, wrapper, *args, **kwargs):
|
||
|
|
with self.__client.cursor() as cursor:
|
||
|
|
data = wrapper(cursor, *args, **kwargs)
|
||
|
|
self.__client.commit()
|
||
|
|
return data
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _fetch_all(cls, cursor: pymysql.cursors.Cursor, sql: str, *args) -> list:
|
||
|
|
#todo research formatting input values in sql queries. Which type of args?
|
||
|
|
cursor.execute(sql, *args)
|
||
|
|
return cursor.fetchall()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _fetch_one(cls, cursor: pymysql.cursors.Cursor, sql: str, *args) -> typing.Any:
|
||
|
|
#todo research formatting input values in sql queries. Which type of args?
|
||
|
|
cursor.execute(sql, *args)
|
||
|
|
return cursor.fetchone()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _execute(cls, cursor: pymysql.cursors.Cursor, sql: str, *args) -> None:
|
||
|
|
#todo research formatting input values in sql queries. Which type of args?
|
||
|
|
cursor.execute(sql, *args)
|
||
|
|
|
||
|
|
def fetch_all(self, sql: str, *args) -> list:
|
||
|
|
return self.__execute_cmd(self._fetch_all, sql, *args)
|
||
|
|
|
||
|
|
def fetch_one(self, sql: str, *args) -> typing.Any | None:
|
||
|
|
return self.__execute_cmd(self._fetch_one, sql, *args)
|
||
|
|
|
||
|
|
def execute(self, sql: str, *args) -> None:
|
||
|
|
return self.__execute_cmd(self._execute, sql, *args)
|