add config and db base

config/config.example.toml

@@ -0,0 +1,6 @@
+[database]
+    db_user="demo_user"
+    db_password="demo_password"
+    db_host="127.0.0.1"
+    db_port=3306
+    db_name="ez_lan_manager"

src/EzLanManager.py

@@ -0,0 +1,17 @@
+import logging
+
+from from_root import from_root
+
+from src.ez_lan_manager.services.ConfigurationService import ConfigurationService
+from src.ez_lan_manager.services.DatabaseService import DatabaseService
+
+from random import randint
+
+logger = logging.getLogger(__name__.split(".")[-1])
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    configuration_service = ConfigurationService(from_root("config.toml"))
+    db_config = configuration_service.get_database_configuration()
+    db_service = DatabaseService(db_config)
+    print(db_service.create_user(f"TestUser{randint(0, 9999)}", f"TestMail{randint(0, 9999)}", "pw123"))

src/ez_lan_manager/__init__.py

@@ -0,0 +1,2 @@
+from src.ez_lan_manager.services import *
+from src.ez_lan_manager.types import *

src/ez_lan_manager/services/ConfigurationService.py

@@ -0,0 +1,31 @@
+import sys
+from pathlib import Path
+import logging
+import tomllib
+
+from src.ez_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
+
+logger = logging.getLogger(__name__.split(".")[-1])
+
+class ConfigurationService:
+    def __init__(self, config_file_path: Path):
+        try:
+            with open(config_file_path, "rb") as config_file:
+                self._config = tomllib.load(config_file)
+        except FileNotFoundError:
+            logger.fatal(f"Could not find config file at \"{config_file_path}\", exiting...")
+            exit(1)
+
+    def get_database_configuration(self) -> DatabaseConfiguration:
+        try:
+            database_configuration = self._config["database"]
+            return DatabaseConfiguration(
+                db_user=database_configuration["db_user"],
+                db_password=database_configuration["db_password"],
+                db_host=database_configuration["db_host"],
+                db_port=database_configuration["db_port"],
+                db_name=database_configuration["db_name"]
+            )
+        except KeyError:
+            logger.fatal("Error loading DatabaseConfiguration, exiting...")
+            sys.exit(1)

src/ez_lan_manager/services/DatabaseService.py

@@ -0,0 +1,92 @@
+import logging
+import sys
+from typing import Optional
+
+import mariadb
+from mariadb import Cursor
+
+from src.ez_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
+from src.ez_lan_manager.types.User import User
+
+logger = logging.getLogger(__name__.split(".")[-1])
+
+class DuplicationError(Exception):
+    pass
+
+class DatabaseService:
+    def __init__(self, database_config: DatabaseConfiguration):
+        self._database_config = database_config
+        try:
+            logger.info(
+                f"Connecting to database '{self._database_config.db_name}' on "
+                f"{self._database_config.db_user}@{self._database_config.db_host}:{self._database_config.db_port}"
+            )
+            self._connection = mariadb.connect(
+                user=self._database_config.db_user,
+                password=self._database_config.db_password,
+                host=self._database_config.db_host,
+                port=self._database_config.db_port,
+                database=self._database_config.db_name
+            )
+        except mariadb.Error as e:
+            logger.fatal(f"Error connecting to database: {e}")
+            sys.exit(1)
+
+    def _get_cursor(self) -> Cursor:
+        return self._connection.cursor()
+
+    @staticmethod
+    def _map_db_result_to_user(data: tuple) -> User:
+        return User(
+            user_id=data[0],
+            user_name=data[1],
+            user_mail=data[2],
+            user_password=data[3],
+            user_first_name=data[4],
+            user_last_name=data[5],
+            user_birth_day=data[6],
+            is_active=bool(data[7]),
+            is_team_member=bool(data[8]),
+            is_admin=bool(data[9]),
+            created_at=data[10],
+            last_updated_at=data[11],
+            balance=int(data[12])
+        )
+
+    def get_user_by_name(self, user_name: str) -> Optional[User]:
+        cursor = self._get_cursor()
+        cursor.execute("SELECT * FROM users WHERE user_name=?", (user_name,))
+        result = cursor.fetchone()
+        if not result:
+            return
+        return self._map_db_result_to_user(result)
+
+    def get_user_by_id(self, user_id: int) -> Optional[User]:
+        cursor = self._get_cursor()
+        cursor.execute("SELECT * FROM users WHERE user_id=?", (user_id,))
+        result = cursor.fetchone()
+        if not result:
+            return
+        return self._map_db_result_to_user(result)
+
+    def get_user_by_main(self, user_mail: str) -> Optional[User]:
+        cursor = self._get_cursor()
+        cursor.execute("SELECT * FROM users WHERE user_mail=?", (user_mail,))
+        result = cursor.fetchone()
+        if not result:
+            return
+        return self._map_db_result_to_user(result)
+
+    def create_user(self, user_name: str, user_mail: str, password_hash: str) -> User:
+        cursor = self._get_cursor()
+        try:
+            cursor.execute(
+                "INSERT INTO users (user_name, user_mail, user_password) "
+                         "VALUES (?, ?, ?)", (user_name, user_mail, password_hash)
+            )
+            self._connection.commit()
+        except mariadb.IntegrityError as e:
+            logger.warning(f"Aborted duplication entry: {e}")
+            raise DuplicationError
+
+        return self.get_user_by_name(user_name)

src/ez_lan_manager/types/ConfigurationTypes.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+@dataclass(frozen=True)
+class DatabaseConfiguration:
+    db_user: str
+    db_password: str
+    db_host: str
+    db_port: int
+    db_name: str

src/ez_lan_manager/types/User.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from datetime import date, datetime
+from typing import Optional
+
+
+@dataclass
+class User:
+    user_id: int
+    user_name: str
+    user_mail: str
+    user_password: str
+    user_first_name: Optional[str]
+    user_last_name: Optional[str]
+    user_birth_day: Optional[date]
+    is_active: bool
+    is_team_member: bool
+    is_admin: bool
+    created_at: datetime
+    last_updated_at: datetime
+    balance: int