117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
import enum
|
|
import os
|
|
|
|
from utils.errors.database import DatabaseConfigError
|
|
from dotenv import load_dotenv
|
|
import logging
|
|
from typing import Any
|
|
|
|
|
|
class DatabaseConfig():
|
|
def __init__(self):
|
|
self.logger = logging.getLogger(__name__)
|
|
self.logger.debug("Parsing .env")
|
|
|
|
anything_set = load_dotenv()
|
|
|
|
if anything_set:
|
|
self.logger.debug(".env found")
|
|
else:
|
|
self.logger.error(".env not found")
|
|
raise FileNotFoundError(".env not found.")
|
|
|
|
host = os.getenv("DATABASE_HOST")
|
|
port = os.getenv("DATABASE_PORT", "3306")
|
|
name = os.getenv("DATABASE_NAME")
|
|
user = os.getenv("DATABASE_USER")
|
|
password = os.getenv("DATABASE_PASSWORD")
|
|
|
|
self.logger.debug("Validating fetched values from .env")
|
|
|
|
if host is None or host == "":
|
|
self.logger.error("DATABASE_HOST is empty")
|
|
raise DatabaseConfigError(
|
|
"Config is invalid.", "DATABASE_HOST", "(Empty)")
|
|
if name is None or name == "":
|
|
self.logger.error("DATABASE_NAME is empty")
|
|
raise DatabaseConfigError(
|
|
"Config is invalid.", "DATABASE_NAME", "(Empty)")
|
|
if user is None or user == "":
|
|
self.logger.error("DATABASE_USER is empty")
|
|
raise DatabaseConfigError(
|
|
"Config is invalid.", "DATABASE_USER", "(Empty)")
|
|
if password is None or password == "":
|
|
self.logger.error("DATABASE_PASSWORD is empty")
|
|
raise DatabaseConfigError(
|
|
"Config is invalid.", "DATABASE_PASSWORD", "(Empty)")
|
|
|
|
if not port.isdigit():
|
|
self.logger.error("DATABASE_PORT is invalid. Not a number")
|
|
raise DatabaseConfigError(
|
|
"Config is invalid", "DATABASE_PORT", port)
|
|
|
|
self.logger.debug("All config validated")
|
|
|
|
self.host = host
|
|
self.port = port
|
|
self.name = name
|
|
self.user = user
|
|
self.password = password
|
|
|
|
|
|
class TransactionLevel(enum.Enum):
|
|
insecure = "READ UNCOMMITTED"
|
|
secure = "SERIALIZABLE"
|
|
|
|
|
|
class UserConfig:
|
|
_instance = None
|
|
|
|
_metadata = {
|
|
"transaction_level": {"friendly_name": "Transaction Level"},
|
|
"simulate_slowdown": {"friendly_name": "Simulate Slowdown"},
|
|
}
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
self._transaction_level = TransactionLevel.insecure
|
|
self._simulate_slowdown = False
|
|
|
|
@property
|
|
def transaction_level(self) -> TransactionLevel:
|
|
return self._transaction_level
|
|
|
|
@transaction_level.setter
|
|
def transaction_level(self, value: Any):
|
|
if not isinstance(value, TransactionLevel):
|
|
raise TypeError(
|
|
f"Invalid value for 'transaction_level'. Must be a TransactionLevel enum, got {type(value).__name__}."
|
|
)
|
|
self._transaction_level = value
|
|
|
|
@property
|
|
def simulate_slowdown(self) -> bool:
|
|
return self._simulate_slowdown
|
|
|
|
@simulate_slowdown.setter
|
|
def simulate_slowdown(self, value: Any):
|
|
if not isinstance(value, bool):
|
|
raise TypeError(
|
|
f"Invalid value for 'simulate_slowdown'. Must be a boolean, got {type(value).__name__}."
|
|
)
|
|
self._simulate_slowdown = value
|
|
|
|
@classmethod
|
|
def get_friendly_name(cls, option: str) -> str:
|
|
return cls._metadata.get(option, {}).get("friendly_name", option)
|
|
|
|
def __dict__(self) -> dict:
|
|
return {
|
|
"transaction_level": self.transaction_level,
|
|
"simulate_slowdown": self.simulate_slowdown,
|
|
}
|