chore: try to make a working base

.gitignore

@@ -1,3 +1,4 @@
+*.db
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[codz]

docs/upgrade_database.md

@@ -0,0 +1,6 @@
+# Alembic
+
+Steps:
+- alembic init alembic
+- change alembic.ini (connection string)
+- change env.py to include base

src/config.ini

@@ -29,7 +29,6 @@ totp_verified = false
 totp_key = ZSI2Z4QHSGVAK6TFVMZXORFDWKHDN4TQ
 
 [Other]
-welcome_session = true
 
 [Database]
 type = sqlite

src/main.py

@@ -2,13 +2,20 @@
 
 import logging as log
 
+import json
+
 from modules.config.reader import reader
+from modules.database.database import database
+from modules.utilities.utilities import utilities as util
 
 if __name__ == '__main__':
     log.basicConfig(level=log.DEBUG)
 
     config_contents = reader.read_config()
-    log.info(config_contents)
-    input()
-    config_contents = reader.refresh_config(config_contents)
-    log.info(config_contents)
+
+    found, config_database = util.filter_config(config_contents, 'DATABASE')
+    if not found:
+        exit(1)
+
+    ok, engine, session = database.create_session(config_database)
+    ok = database.verify_contents(engine)

src/modules/config/reader.py

@@ -2,23 +2,25 @@
 
 import logging as log
 
-from .utilities import preflight_checks
+from .utilities import checks
 
 class reader():
+    @staticmethod
     def read_config() -> dict:
         '''
         check some basic things and then return the dict containing the config data
         '''
 
-        ok, candidate_path = preflight_checks.search_known_paths()
+        ok, candidate_path = checks.search_known_paths()
         if not ok:
             return {}
-        ok, config_contents = preflight_checks.verify_contents(candidate_path)
+        ok, config_contents = checks.verify_contents(candidate_path)
         if not ok:
             return {}
 
         return config_contents
 
+    @staticmethod
     def refresh_config(config_contents: dict) -> dict:
         log.debug(f'refreshing config values')
         return reader.read_config()

src/modules/config/utilities.py

@@ -5,7 +5,8 @@ import logging as log
 import configparser as cp
 import os
 
-class preflight_checks():
+class checks():
+    @staticmethod
     def search_known_paths() -> tuple[bool, str]:
         '''
         Look at predefined paths on the filesystem for a config file
@@ -34,6 +35,7 @@ class preflight_checks():
             log.error(f'error occured while searching for the config file: {err}')
             return False, ''
 
+    @staticmethod
     def verify_contents(config_path: str) -> tuple[bool, dict]:
         '''
         Check the existing config file for contents
@@ -53,10 +55,15 @@ class preflight_checks():
                 log.debug(f'checking integrity of section: {section}')
 
                 if len(config.items(section)) == 0:
-                    log.error('empty section, no keys or values')
-                    return False, {}
+                    log.warn('empty section, removing for runtime due to irrelevance')
+                    config.remove_section(section)
 
-            return True, dict(config.items())
+            config_dict = {}
+            for section in config.sections():
+                items = dict(config.items(section))
+                config_dict[section] = items
+
+            return True, config_dict
 
         except cp.ParsingError as err:
             log.error(f'error parsing the ini config file: {err}')

src/modules/database/database.py

@@ -1,2 +1,43 @@
 #!/bin/env python3
 
+import logging as log
+
+import sqlalchemy
+import sqlalchemy.orm
+
+from .schema import Base
+from .utilities import checks
+
+class database():
+    @staticmethod
+    def create_session(database_config: dict) -> tuple[bool, sqlalchemy.engine.Engine | None, sqlalchemy.orm.Session | None]:
+        ok, connection_string = checks.generate_connection_string(database_config)
+        if not ok:
+            return False, None, None
+
+        log.info(connection_string)
+
+        try:
+            engine = sqlalchemy.create_engine(connection_string, echo=False)
+
+            local_session = sqlalchemy.orm.sessionmaker(bind=engine)
+            session = local_session()
+
+            return True, engine, session
+
+        except Exception as err:
+            log.critical(f'database initialization failed: {err}')
+            return False, None, None
+    
+    @staticmethod
+    def verify_contents(engine) -> bool:
+        try:
+            log.info('checking if all tables are present, and creating them if they are not')
+
+            Base.metadata.create_all(engine)
+
+            return True
+
+        except Exception as err:
+            log.error('failed to verify contents of the database')
+            return False

src/modules/database/schema.py

@@ -1,2 +1,25 @@
 #!/bin/env python3
 
+import sqlalchemy
+import sqlalchemy.orm
+
+Base = sqlalchemy.orm.declarative_base()
+
+class User(Base):
+    __tablename__ = 'users'
+
+    user_id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, index=True)
+
+    username = sqlalchemy.Column(sqlalchemy.String, unique=True, index=True)
+    password = sqlalchemy.Column(sqlalchemy.String)
+
+    role = sqlalchemy.Column(sqlalchemy.String, default='user')
+
+    totp_enabled = sqlalchemy.Column(sqlalchemy.Boolean, default=False)
+    totp_verified = sqlalchemy.Column(sqlalchemy.Boolean, default=False)
+    totp_key = sqlalchemy.Column(sqlalchemy.String)
+
+    email = sqlalchemy.Column(sqlalchemy.String)
+
+class Wireguard(Base):
+    __tablename__ = 'wireguard_interfaces'

src/modules/database/utilities.py

@@ -1,9 +1,45 @@
 #!/bin/env python3
 
-class db_setup():
+import logging as log
+
+import os
+from urllib.parse import quote_plus
+
+from modules.utilities.utilities import utilities as util
+
+class checks():
     '''
     This class functions as a collection of function to prepare a connection to a database.
     '''
 
-    def compile_connection_string() -> string:
-        return "The Connection string"
+    @staticmethod
+    def generate_connection_string(database_config: dict) -> tuple[bool, str]:
+        if not 'type' in database_config:
+            return False, ''
+
+        username = quote_plus(database_config.get('username', ''))
+        password = quote_plus(database_config.get('password', ''))
+
+        match database_config['type']:
+            case 'sqlite':
+                local_database_path = os.path.abspath("./database")
+
+                exists = util.ensure_directory(local_database_path)
+                if exists:
+                    connection_string = f'sqlite:///{local_database_path}/wgdashboard.db'
+                else:
+                    return False, ''
+
+            case 'postgresql' | 'postgres':
+                connection_string = f'postgresql+psycopg://{username}:{password}@{database_config.get('host', 'localhost')}:{database_config.get('port', '5432')}'
+
+            case 'mariadb':
+                connection_string = f'mariadb+mariadbconnector://{username}:{password}@{database_config.get('host', 'localhost')}:{database_config.get('port', '3306')}'
+
+            case 'mysql':
+                connection_string = f'mysql+pymysql://{username}:{password}@{database_config.get('host', 'localhost')}:{database_config.get('port', '3306')}'
+
+            case _:
+                return False, ''
+
+        return True, connection_string

src/modules/utilities/utilities.py

@@ -0,0 +1,39 @@
+#!/bin/env python3
+
+import logging as log
+
+import os
+
+class utilities():
+    @staticmethod
+    def filter_config(config_contents: dict, filter_keyword: str) -> tuple[bool, dict]:
+        '''
+        Helper function to grab a specific part of the config
+        '''
+
+        log.debug(f'searching for section: {filter_keyword}')
+        for section in config_contents:
+            if str(section).lower() == filter_keyword.lower():
+                return True, dict(config_contents[section].items())
+
+        return False, {}
+
+    @staticmethod
+    def ensure_directory(path: str) -> bool:
+        '''
+        Make the directory if it does not exist yet, return only true if the directory was missing and created.
+        '''
+
+        log.debug(f'checking if the directory at: {path} exists')
+        if os.path.exists(path) and os.path.isdir(path):
+            return True
+        
+        try:
+            os.mkdir(path)
+            return True
+
+        except Exception as err:
+            log.critical('failed to create directory')
+            return False
+        
+        

src/requirements.txt

@@ -1 +1,2 @@
 configparser==7.2.0
+sqlalchemy==2.0.47