feat: Convert config template to pydantic model

backend/.gitignore

@@ -3,7 +3,6 @@
 
 .idea
 config/*
-!config/config_template.yaml
 .history
 **/__pycache__/
 .vscode

backend/capellacollab/__init__.py

@@ -3,13 +3,8 @@
 
 from importlib import metadata
 
-from capellacollab import config as config_module
-
 try:
     __version__ = metadata.version("capellacollab-backend")
 except metadata.PackageNotFoundError:
     __version__ = "0.0.0+unknown"
 del metadata
-
-
-config_module.validate_schema()

backend/capellacollab/__main__.py

@@ -50,18 +50,18 @@
 handlers: list[logging.Handler] = [
     logging.StreamHandler(),
     core_logging.CustomTimedRotatingFileHandler(
-        str(config["logging"]["logPath"]) + "backend.log"
+        str(config.logging.logPath) + "backend.log"
     ),
 ]
 
 for handler in handlers:
     handler.setFormatter(core_logging.CustomFormatter())
 
-logging.basicConfig(level=config["logging"]["level"], handlers=handlers)
+logging.basicConfig(level=config.logging.level, handlers=handlers)
 
 
 async def startup():
-    migration.migrate_db(engine, config["database"]["url"])
+    migration.migrate_db(engine, config.database.url)
     logging.info("Migrations done - Server is running")
 
     # This is needed to load the Kubernetes configuration at startup

backend/capellacollab/alembic/env.py

@@ -16,7 +16,7 @@
 # access to the values within the .ini file in use.
 config = context.config
 
-logging.basicConfig(level=cfg["logging"]["level"])
+logging.basicConfig(level=cfg.logging.level)
 if os.getenv("ALEMBIC_CONFIGURE_LOGGER", "true") != "false":
     logging.getLogger("capellacollab").setLevel("WARNING")
 
@@ -25,7 +25,7 @@
 # this will overwrite the ini-file sqlalchemy.url path
 # with the path given in the config of the main code
 if not config.get_main_option("sqlalchemy.url"):
-    config.set_main_option("sqlalchemy.url", cfg["database"]["url"])
+    config.set_main_option("sqlalchemy.url", cfg.database.url)
 
 # Import models
 

backend/capellacollab/config/__init__.py

@@ -1,23 +1,18 @@
 # SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
 # SPDX-License-Identifier: Apache-2.0
 
-
 import logging
 
-import jsonschema
-import jsonschema.exceptions
-
-from . import exceptions, loader
+from . import generate, loader, models
 
 log = logging.getLogger(__name__)
-config = loader.load_yaml()
+
+if not loader.does_config_exist():
+    log.warning(
+        "No configuration file found. Generating default configuration at backend/capellacollab/config/config.yaml"
+    )
+    generate.write_config()
 
 
-def validate_schema():
-    config_schema = loader.load_config_schema()
-    try:
-        jsonschema.validate(config, config_schema)
-    except jsonschema.exceptions.ValidationError as error:
-        raise exceptions.InvalidConfigurationError(
-            f"{error.__class__.__name__}: {error.message}",
-        ) from None
+config_data = loader.load_yaml()
+config = models.AppConfig(**config_data)

backend/capellacollab/config/diff.py

@@ -3,12 +3,10 @@
 
 # pylint: disable=bad-builtin
 
-import pathlib
-
 import deepdiff
-import yaml
 
 from . import loader
+from . import models as config_models
 
 
 class bcolors:
@@ -22,16 +20,15 @@ class bcolors:
 
 print("Start comparison of configuration files")
 
-config_template = yaml.safe_load(
-    (
-        pathlib.Path(__file__).parents[2] / "config" / "config_template.yaml"
-    ).open()
-)
+pydantic_config = config_models.AppConfig()
 
 config = loader.load_yaml()
 
 diff = deepdiff.DeepDiff(
-    config, config_template, ignore_order=True, report_repetition=True
+    config,
+    pydantic_config.model_dump(),
+    ignore_order=True,
+    report_repetition=True,
 )
 
 for key, value in (

backend/capellacollab/config/generate.py

@@ -0,0 +1,20 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+
+import yaml
+
+from . import models
+
+
+def write_config():
+    current_dir = os.path.dirname(os.path.realpath(__file__))
+    config_path = os.path.join(current_dir, "config.yaml")
+
+    app_config = models.AppConfig()
+    config_dict = app_config.model_dump()
+    yaml_str = yaml.dump(config_dict, sort_keys=False)
+
+    with open(config_path, "w", encoding="utf-8") as yaml_file:
+        yaml_file.write(yaml_str)

backend/capellacollab/config/loader.py

@@ -11,12 +11,14 @@
 from . import exceptions
 
 log = logging.getLogger(__name__)
+CONFIG_FILE_NAME = "config.yaml"
 
 config_locations: list[pathlib.Path] = [
-    pathlib.Path(__file__).parents[2] / "config" / "config.yaml",
+    pathlib.Path(__file__).parents[0] / CONFIG_FILE_NAME,
+    pathlib.Path(__file__).parents[2] / "config" / CONFIG_FILE_NAME,
     pathlib.Path(appdirs.user_config_dir("capellacollab", "db"))
-    / "config.yaml",
-    pathlib.Path("/etc/capellacollab") / "config.yaml",
+    / CONFIG_FILE_NAME,
+    pathlib.Path("/etc/capellacollab") / CONFIG_FILE_NAME,
 ]
 
 config_fallback_locations: list[pathlib.Path] = [
@@ -38,12 +40,24 @@ def construct_mapping(self, node, deep=False):
         return super().construct_mapping(node, deep)
 
 
+def does_config_exist() -> bool:
+    for loc in config_locations:
+        if loc.exists():
+            return True
+
+    for loc in config_fallback_locations:
+        if loc.exists():
+            return True
+
+    return False
+
+
 def load_yaml() -> dict:
     log.debug("Searching for configuration files...")
     for loc in config_locations:
         if loc.exists():
             log.info("Loading configuration file at location %s", str(loc))
-            return yaml.load(loc.open(), UniqueKeyLoader)
+            return yaml.load(loc.open(encoding="utf-8"), UniqueKeyLoader)
         else:
             log.debug(
                 "Didn't find a configuration file at location %s", str(loc)
@@ -54,7 +68,7 @@ def load_yaml() -> dict:
             log.warning(
                 "Loading fallback configuration file at location %s", str(loc)
             )
-            return yaml.safe_load(loc.open())
+            return yaml.safe_load(loc.open(encoding="utf-8"))
 
     raise FileNotFoundError("config.yaml")
 

backend/capellacollab/config/models.py

@@ -0,0 +1,158 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+import pydantic
+
+
+class DockerConfig(pydantic.BaseModel):
+    registry: str = "k3d-myregistry.localhost:12345"
+    externalRegistry: str = "docker.io"
+
+
+class K8sPodSecurityContext(pydantic.BaseModel):
+    runAsUser: int = 1004370000
+    runAsGroup: int = 1004370000
+    fsGroup: int = 1004370000
+    runAsNonRoot: bool = True
+
+
+class K8sClusterConfig(pydantic.BaseModel):
+    imagePullPolicy: str = "Always"
+    podSecurityContext: K8sPodSecurityContext = K8sPodSecurityContext()
+
+
+class K8sPromtailConfig(pydantic.BaseModel):
+    lokiEnabled: bool = True
+    lokiUrl: str = "http://localhost:30001/loki/api/v1/push"
+    lokiUsername: str = "localLokiUser"
+    lokiPassword: str = "localLokiPassword"
+    serverPort: int = 3101
+
+
+# Only required when using operator k8s
+
+
+class K8sConfig(pydantic.BaseModel):
+    # Only required, if you'd like to use a local k3d environment
+    context: str = "k3d-collab-cluster"
+    namespace: str = "collab-sessions"
+    storageClassName: str = "local-path"
+    storageAccessMode: str = "ReadWriteOnce"
+    cluster: K8sClusterConfig = K8sClusterConfig()
+    promtail: K8sPromtailConfig = K8sPromtailConfig()
+    ingressClassName: str = "traefik"
+    # Only required when no kubectl context is available
+    # apiURL: str | None = None
+    # token: str | None = None
+
+
+class GeneralConfig(pydantic.BaseModel):
+    host: str = "localhost"
+    port: int = 8000
+    scheme: str = "http"
+    wildcardHost: bool = False
+
+
+class ExtensionGuacamoleConfig(pydantic.BaseModel):
+    baseURI: str = "http://localhost:8080/guacamole"
+    publicURI: str = "http://localhost:8080/guacamole"
+    username: str = "guacadmin"
+    password: str = "guacadmin"
+
+
+class ExtensionJupyterConfig(pydantic.BaseModel):
+    publicURI: str = "http://localhost:8080/jupyter"
+
+
+class ExtensionsConfig(pydantic.BaseModel):
+    guacamole: ExtensionGuacamoleConfig = ExtensionGuacamoleConfig()
+    jupyter: ExtensionJupyterConfig = ExtensionJupyterConfig()
+
+
+class AuthOauthClientConfig(pydantic.BaseModel):
+    id: str = "default"
+    secret: str | None = None
+
+
+# Only required when using provider oauth
+
+
+class AuthOathEndpointsConfig(pydantic.BaseModel):
+    wellKnown: str = (
+        "http://localhost:8083/default/.well-known/openid-configuration"
+    )
+    tokenIssuance: str | None = None
+    authorization: str | None = None
+
+
+class AuthOauthConfig(pydantic.BaseModel):
+    # Only required when using provider oauth
+    endpoints: AuthOathEndpointsConfig = AuthOathEndpointsConfig()
+    audience: str = "default"
+    scopes: list[str] = ["openid"]
+    client: AuthOauthClientConfig = AuthOauthClientConfig()
+    redirectURI: str = "http://localhost:4200/oauth2/callback"
+
+
+class JWTConfig(pydantic.BaseModel):
+    usernameClaim: str = "sub"  # preferred_username
+
+
+class AzureClientConfig(pydantic.BaseModel):
+    id: str = "tbd"
+    secret: str = "tbd"
+
+
+class AzureConfig(pydantic.BaseModel):
+    authorizationEndpoint: str = "tbd"
+    audience: str = "tbd"
+    client: AzureClientConfig = AzureClientConfig()
+    redirectURI: str = "http://localhost:4200/oauth2/callback"
+
+
+class AuthenticationConfig(pydantic.BaseModel):
+    provider: str = "oauth"  # oauth | azure
+    jwt: JWTConfig = JWTConfig()
+    oauth: AuthOauthConfig = AuthOauthConfig()
+    azure: AzureConfig = (
+        AzureConfig()
+    )  # Only required when using provider azure
+
+
+class PipelineConfig(pydantic.BaseModel):
+    timeout: int = 60
+
+
+class DatabaseConfig(pydantic.BaseModel):
+    url: str = "postgresql://dev:dev@localhost:5432/dev"
+
+
+class InitialConfig(pydantic.BaseModel):
+    admin: str = "admin"
+
+
+class LoggingConfig(pydantic.BaseModel):
+    level: str = "DEBUG"
+    logPath: str = "logs/"
+
+
+class RequestsConfig(pydantic.BaseModel):
+    timeout: int = 2
+
+
+class PrometheusConfig(pydantic.BaseModel):
+    url: str = "http://localhost:8080/prometheus/"
+
+
+class AppConfig(pydantic.BaseModel):
+    docker: DockerConfig = DockerConfig()
+    k8s: K8sConfig = K8sConfig()
+    general: GeneralConfig = GeneralConfig()
+    extensions: ExtensionsConfig = ExtensionsConfig()
+    authentication: AuthenticationConfig = AuthenticationConfig()
+    pipelines: PipelineConfig = PipelineConfig()
+    database: DatabaseConfig = DatabaseConfig()
+    initial: InitialConfig = InitialConfig()
+    logging: LoggingConfig = LoggingConfig()
+    requests: RequestsConfig = RequestsConfig()
+    prometheus: PrometheusConfig = PrometheusConfig()

backend/capellacollab/core/authentication/__init__.py

@@ -14,11 +14,10 @@ def get_authentication_entrypoint():
             for i in metadata.entry_points()[
                 "capellacollab.authentication.providers"
             ]
-            if i.name == config["authentication"]["provider"]
+            if i.name == config.authentication.provider
         )
         return ep
     except StopIteration:
         raise ValueError(
-            "Unknown authentication provider "
-            + config["authentication"]["provider"]
+            "Unknown authentication provider " + config.authentication.provider
         ) from None

backend/capellacollab/core/authentication/jwt_bearer.py

@@ -52,9 +52,7 @@ async def __call__(  # type: ignore
         return None
 
     def get_username(self, token_decoded: dict[str, str]) -> str:
-        return token_decoded[
-            config["authentication"]["jwt"]["usernameClaim"]
-        ].strip()
+        return token_decoded[config.authentication.jwt.usernameClaim].strip()
 
     def initialize_user(self, token_decoded: dict[str, str]):
         with database.SessionLocal() as session: