Improved project file structure

tests/custom/database/src/application.py

@@ -0,0 +1,38 @@
+from typing import Optional
+
+from cpl_core.application import ApplicationABC
+from cpl_core.configuration import ConfigurationABC
+from cpl_core.console import Console
+from cpl_core.dependency_injection import ServiceProviderABC
+from cpl_core.logging import LoggerABC
+from model.user_repo_abc import UserRepoABC
+from model.user_repo import UserRepo
+
+
+class Application(ApplicationABC):
+
+    def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
+        ApplicationABC.__init__(self, config, services)
+
+        self._logger: Optional[LoggerABC] = None
+
+    def configure(self):
+        self._logger = self._services.get_service(LoggerABC)
+
+    def main(self):
+        self._logger.header(f'{self._configuration.environment.application_name}:')
+        self._logger.debug(__name__, f'Host: {self._configuration.environment.host_name}')
+        self._logger.debug(__name__, f'Environment: {self._configuration.environment.environment_name}')
+        self._logger.debug(__name__, f'Customer: {self._configuration.environment.customer}')
+
+        user_repo: UserRepo = self._services.get_service(UserRepoABC)
+        if len(user_repo.get_users()) == 0:
+            user_repo.add_test_user()
+        
+        Console.write_line('Users:')
+        for user in user_repo.get_users():
+            Console.write_line(user.UserId, user.Name, user.City)
+
+        Console.write_line('Cities:')
+        for city in user_repo.get_cities():
+            Console.write_line(city.CityId, city.Name, city.ZIP)

tests/custom/database/src/appsettings.development.json

@@ -0,0 +1,8 @@
+{
+  "LoggingSettings": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLogLevel": "TRACE",
+    "FileLogLevel": "TRACE"
+  }
+}

tests/custom/database/src/appsettings.edrafts-lapi.json

@@ -0,0 +1,22 @@
+{
+  "TimeFormatSettings": {
+    "DateFormat": "%Y-%m-%d",
+    "TimeFormat": "%H:%M:%S",
+    "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
+    "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
+  },
+
+  "LoggingSettings": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLogLevel": "TRACE",
+    "FileLogLevel": "TRACE"
+  },
+
+  "DatabaseSettings": {
+    "AuthPlugin": "mysql_native_password",
+    "ConnectionString": "mysql+mysqlconnector://sh_cpl:$credentials@localhost/sh_cpl",
+    "Credentials": "MHZhc0Y2bjhKc1VUMWV0Qw==",
+    "Encoding": "utf8mb4"
+  }
+}

tests/custom/database/src/appsettings.edrafts-pc.json

@@ -0,0 +1,26 @@
+{
+  "TimeFormatSettings": {
+    "DateFormat": "%Y-%m-%d",
+    "TimeFormat": "%H:%M:%S",
+    "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
+    "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
+  },
+
+  "LoggingSettings": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLogLevel": "TRACE",
+    "FileLogLevel": "TRACE"
+  },
+
+  "DatabaseSettings": {
+    "Host": "localhost",
+    "User": "sh_cpl",
+    "Password": "MHZhc0Y2bjhKc1VUMWV0Qw==",
+    "Database": "sh_cpl",
+    "Charset": "utf8mb4",
+    "UseUnicode": "true",
+    "Buffered": "true",
+    "AuthPlugin": "mysql_native_password"
+  }
+}

tests/custom/database/src/appsettings.json

@@ -0,0 +1,15 @@
+{
+  "TimeFormatSettings": {
+    "DateFormat": "%Y-%m-%d",
+    "TimeFormat": "%H:%M:%S",
+    "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
+    "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
+  },
+
+  "LoggingSettings": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLogLevel": "ERROR",
+    "FileLogLevel": "WARN"
+  }
+}

tests/custom/database/src/main.py

@@ -0,0 +1,14 @@
+from cpl_core.application import ApplicationBuilder
+
+from application import Application
+from startup import Startup
+
+
+def main():
+    app_builder = ApplicationBuilder(Application)
+    app_builder.use_startup(Startup)
+    app_builder.build().run()
+
+
+if __name__ == '__main__':
+    main()

tests/custom/database/src/model/city_model.py

@@ -0,0 +1,47 @@
+from cpl_core.database import TableABC
+
+
+class CityModel(TableABC):
+
+    def __init__(self, name: str, zip_code: str, id = 0):
+        self.CityId = id
+        self.Name = name
+        self.ZIP = zip_code
+    
+    @staticmethod
+    def get_create_string() -> str:
+        return str(f"""
+            CREATE TABLE IF NOT EXISTS `City` (
+                `CityId` INT(30) NOT NULL AUTO_INCREMENT,
+                `Name` VARCHAR(64) NOT NULL,
+                `ZIP` VARCHAR(5) NOT NULL,
+                PRIMARY KEY(`CityId`)
+            );
+        """)
+    
+    @property
+    def insert_string(self) -> str:
+        return str(f"""
+            INSERT INTO `City` (
+                `Name`, `ZIP`
+            ) VALUES (
+                '{self.Name}',
+                '{self.ZIP}'
+            );
+        """)
+    
+    @property
+    def udpate_string(self) -> str:
+        return str(f"""
+            UPDATE `City`
+            SET `Name` = '{self.Name}', 
+            `ZIP` = '{self.ZIP}',
+            WHERE `CityId` = {self.Id};
+        """)
+    
+    @property
+    def delete_string(self) -> str:
+        return str(f"""
+            DELETE FROM `City`
+            WHERE `CityId` = {self.Id};
+        """)

tests/custom/database/src/model/db_context.py

@@ -0,0 +1,8 @@
+from cpl_core.database import DatabaseSettings
+from cpl_core.database.context import DatabaseContext
+
+
+class DBContext(DatabaseContext):
+
+    def __init__(self, db_settings: DatabaseSettings):
+        DatabaseContext.__init__(self, db_settings)

tests/custom/database/src/model/user_model.py

@@ -0,0 +1,50 @@
+from cpl_core.database import TableABC
+
+from .city_model import CityModel
+
+
+class UserModel(TableABC):
+
+    def __init__(self, name: str, city: CityModel, id = 0):
+        self.UserId = id
+        self.Name = name
+        self.CityId = city.CityId if city is not None else 0
+        self.City = city
+
+    @staticmethod
+    def get_create_string() -> str:
+        return str(f"""
+            CREATE TABLE IF NOT EXISTS `User` (
+                `UserId` INT(30) NOT NULL AUTO_INCREMENT,
+                `Name` VARCHAR(64) NOT NULL,
+                `CityId` INT(30),
+                FOREIGN KEY (`UserId`) REFERENCES City(`CityId`),
+                PRIMARY KEY(`UserId`)
+            );
+        """)
+
+    @property
+    def insert_string(self) -> str:
+        return str(f"""
+            INSERT INTO `User` (
+                `Name`
+            ) VALUES (
+                '{self.Name}'
+            );
+        """)
+
+    @property
+    def udpate_string(self) -> str:
+        return str(f"""
+            UPDATE `User`
+            SET `Name` = '{self.Name}', 
+            `CityId` = {self.CityId},
+            WHERE `UserId` = {self.UserId};
+        """)
+
+    @property
+    def delete_string(self) -> str:
+        return str(f"""
+            DELETE FROM `User`
+            WHERE `UserId` = {self.UserId};
+        """)

tests/custom/database/src/model/user_repo.py

@@ -0,0 +1,42 @@
+from cpl_core.console import Console
+from cpl_core.database.context import DatabaseContextABC
+
+from .city_model import CityModel
+from .user_model import UserModel
+from .user_repo_abc import UserRepoABC
+
+
+class UserRepo(UserRepoABC):
+
+    def __init__(self, db_context: DatabaseContextABC):
+        UserRepoABC.__init__(self)
+
+        self._db_context: DatabaseContextABC = db_context
+
+    def add_test_user(self):
+        city = CityModel('Haren', '49733')
+        city2 = CityModel('Meppen', '49716')
+        self._db_context.cursor.execute(city2.insert_string)
+        user = UserModel('TestUser', city)
+        self._db_context.cursor.execute(user.insert_string)
+        self._db_context.save_changes()
+
+    def get_users(self) -> list[UserModel]:
+        users =  []
+        results = self._db_context.select('SELECT * FROM `User`')
+        for result in results:
+            users.append(UserModel(result[1], self.get_city_by_id(result[2]), id = result[0]))
+        return users
+
+    def get_cities(self) -> list[CityModel]:
+        cities =  []
+        results = self._db_context.select('SELECT * FROM `City`')
+        for result in results:
+            cities.append(CityModel(result[1], result[2], id = result[0]))
+        return cities
+    
+    def get_city_by_id(self, id: int) -> CityModel:
+        if id is None:
+            return None
+        result = self._db_context.select(f'SELECT * FROM `City` WHERE `Id` = {id}')
+        return CityModel(result[1], result[2], id = result[0])

tests/custom/database/src/model/user_repo_abc.py

@@ -0,0 +1,19 @@
+from abc import ABC, abstractmethod
+
+from .city_model import CityModel
+from .user_model import UserModel
+
+
+class UserRepoABC(ABC):
+
+    @abstractmethod
+    def __init__(self): pass
+
+    @abstractmethod
+    def get_users(self) -> list[UserModel]: pass
+    
+    @abstractmethod
+    def get_cities(self) -> list[CityModel]: pass
+    
+    @abstractmethod
+    def get_city_by_id(self, id: int) -> CityModel: pass

tests/custom/database/src/startup.py

@@ -0,0 +1,41 @@
+from cpl_core.application import StartupABC
+from cpl_core.configuration import ConfigurationABC
+from cpl_core.database import DatabaseSettings
+from cpl_core.dependency_injection import (ServiceCollectionABC,
+                                           ServiceProviderABC)
+from cpl_core.environment import ApplicationEnvironmentABC
+from cpl_core.logging import Logger, LoggerABC
+
+from model.db_context import DBContext
+from model.user_repo import UserRepo
+from model.user_repo_abc import UserRepoABC
+
+
+class Startup(StartupABC):
+
+    def __init__(self):
+        StartupABC.__init__(self)
+        
+        self._configuration = None
+
+    def configure_configuration(self, configuration: ConfigurationABC, environment: ApplicationEnvironmentABC) -> ConfigurationABC:
+        configuration.add_environment_variables('PYTHON_')
+        configuration.add_environment_variables('CPL_')
+        configuration.parse_console_arguments()
+        configuration.add_json_file(f'appsettings.json')
+        configuration.add_json_file(f'appsettings.{configuration.environment.environment_name}.json')
+        configuration.add_json_file(f'appsettings.{configuration.environment.host_name}.json', optional=True)
+
+        self._configuration = configuration
+        
+        return configuration
+
+    def configure_services(self, services: ServiceCollectionABC, environment: ApplicationEnvironmentABC) -> ServiceProviderABC:
+        # Create and connect to database
+        db_settings: DatabaseSettings = self._configuration.get_configuration(DatabaseSettings)
+        services.add_db_context(DBContext, db_settings)
+
+        services.add_singleton(UserRepoABC, UserRepo)
+
+        services.add_singleton(LoggerABC, Logger)
+        return services.build_service_provider()

tests/custom/database/src/tests/__init__.py

@@ -0,0 +1 @@
+# imports: