Almost all requested changes

Author: mephenor

src/ets/core/aem_pack_registry.py

@@ -30,11 +30,11 @@
 from schemapack.spec.schemapack import SchemaPack
 
 from ets.config import Config
+from ets.core.config_manager import ConfigManager
 from ets.core.models import AEMPack, PersistedConfig, Workflow
 from ets.ports.inbound.aem_pack_registry import (
     AEMPackRegistryPort,
 )
-from ets.ports.inbound.config_manager import ConfigManagerPort
 from ets.ports.outbound.config_lock import ConfigLockPort
 from ets.ports.outbound.dao import AEMPackDao
 from ets.ports.outbound.incoming_aem_pack_queue import IncomingAEMPackQueuePort
@@ -56,7 +56,7 @@ def __init__(
         *,
         config: Config,
         aem_pack_dao: AEMPackDao,
-        config_manager: ConfigManagerPort,
+        config_manager: ConfigManager,
         config_lock: ConfigLockPort,
         incoming_aem_pack_queue: IncomingAEMPackQueuePort,
     ):

src/ets/core/config_manager.py

@@ -21,17 +21,20 @@
 from ets.core.config_comparison import compare_configs
 from ets.core.config_pruning import prune_unproductive_subgraphs
 from ets.core.config_validation import ConfigValidationError, validate
-from ets.core.model_derivation import ModelDeriver
+from ets.core.model_derivation import ModelDerivationError, ModelDeriver
 from ets.core.models import PersistedConfig, RawConfig
-from ets.ports.inbound.config_manager import ConfigManagerError, ConfigManagerPort
 from ets.ports.outbound.config_loader import ConfigLoaderPort
 from ets.ports.outbound.config_version import ConfigVersionerPort
 from ets.ports.outbound.config_writer import ConfigWriterPort
 
 log = logging.getLogger(__name__)
 
 
-class ConfigManager(ConfigManagerPort):
+class ConfigManagerError(RuntimeError):
+    """Raised when an unexpected error happens while handling a transformation configuration."""
+
+
+class ConfigManager:
     """Manages loading, comparison, validation and selection of an active config."""
 
     def __init__(
@@ -64,7 +67,7 @@ def known_version(self) -> int:
         return self._known_version
 
     async def update_config(self):
-        """Return the active config and its version, reloading from DB if the version changed."""
+        """Check if the active config is stale and reload from DB if the version changed."""
         current_version = await self._versioner.get_version()
         if self._current_config is None:
             log.info("Loading initial config (version %d).", current_version)
@@ -86,7 +89,7 @@ async def update_config(self):
             log.critical(inconsistent_version)
             raise inconsistent_version
 
-    async def resolve_and_persist(self, input_config_path: Path) -> None:
+    async def resolve_and_persist(self, input_config_path: Path) -> bool:
         """Load, resolve, and persist the transformation config.
 
         - Loads the raw config from disk and the persisted one from the database.
@@ -100,19 +103,28 @@ async def resolve_and_persist(self, input_config_path: Path) -> None:
         Raises:
             ConfigManagerError: If the new config fails validation and no previous
                 valid config exists in the database.
+        Returns:
+            True, if the persisted config is outdated and has been replaced with a new one
+            False in all other cases
         """
         raw_config = self._loader.load_config_from_file(input_config_path)
         persisted_config = await self._loader.load_config_from_db()
 
-        await self._resolve(raw_config=raw_config, persisted_config=persisted_config)
+        return await self._resolve_has_config_changed(
+            raw_config=raw_config, persisted_config=persisted_config
+        )
 
-    async def _resolve(
+    async def _resolve_has_config_changed(
         self, *, raw_config: RawConfig, persisted_config: PersistedConfig
-    ):
-        """Compare, validate/prune and derive schemas as needed."""
+    ) -> bool:
+        """Compare, validate/prune, derive schemas and persist the config as needed.
+        Returns:
+            True, if the persisted config is outdated and has been replaced with a new one
+            False in all other cases
+        """
         result = compare_configs(raw_config, persisted_config)
         if not isinstance(result, RawConfig):
-            return
+            return False
 
         try:
             validated = validate(raw_config)
@@ -134,11 +146,21 @@ async def _resolve(
             log.warning(
                 "New config failed to validate, using existing, persisted config instead."
             )
-            return persisted_config
-        derived_models = self._model_deriver.derive_models(pruned)
+            return False
+
+        try:
+            derived_models = self._model_deriver.derive_models(pruned)
+        except ModelDerivationError:
+            log.warning(
+                "Could not derive model schemas for new configuration."
+                + "\nFalling back to existing, persisted config instead."
+            )
+            return False
+
         resolved = PersistedConfig(
             models=derived_models,
             routes=pruned.routes,
             workflows=pruned.workflows,
         )
         await self._writer.write_config(resolved)
+        return True

src/ets/core/config_updater.py

@@ -18,31 +18,27 @@
 import logging
 from pathlib import Path
 
-from ets.ports.inbound.config_manager import ConfigManagerPort
-from ets.ports.inbound.config_updater import ConfigUpdaterPort
+from ets.core.config_manager import ConfigManager
 from ets.ports.outbound.config_lock import ConfigLockPort
-from ets.ports.outbound.config_version import ConfigVersionerPort
 from ets.ports.outbound.incoming_aem_pack_queue import IncomingAEMPackQueuePort
 
 log = logging.getLogger(__name__)
 
 
-class ConfigUpdater(ConfigUpdaterPort):
+class ConfigUpdater:
     """Coordinates the startup config update across service instances."""
 
     def __init__(
         self,
         *,
         input_config_path: Path,
         config_lock: ConfigLockPort,
-        config_manager: ConfigManagerPort,
-        versioner: ConfigVersionerPort,
+        config_manager: ConfigManager,
         incoming_aem_pack_queue: IncomingAEMPackQueuePort,
     ):
         self._input_config_path = input_config_path
         self._config_lock = config_lock
         self._config_manager = config_manager
-        self._versioner = versioner
         self._incoming_aem_pack_queue = incoming_aem_pack_queue
 
     async def run(self) -> None:
@@ -57,15 +53,15 @@ async def run(self) -> None:
         await self._config_lock.setup_index()
         if not await self._config_lock.try_acquire_lock():
             await self._config_lock.wait_for_lock_release()
-            log.info("Update lock released, loading persisted config placeholder.")
+            log.info("Update lock released, loading persisted config.")
             return
 
         try:
             log.info("Lock acquired, starting config update.")
-            previous_version = await self._versioner.get_version()
-            await self._config_manager.resolve_and_persist(self._input_config_path)
-            current_version = await self._versioner.get_version()
-            if current_version != previous_version:
+            config_has_changed = await self._config_manager.resolve_and_persist(
+                self._input_config_path
+            )
+            if config_has_changed:
                 await self._incoming_aem_pack_queue.mark_all_for_reprocessing()
             log.info("Config validation/update finished.")
         finally:

src/ets/core/config_validation.py

@@ -87,7 +87,7 @@ def _validate_routes(raw_config: RawConfig) -> None:
         if output_model.is_ingress:
             raise ConfigValidationError(
                 f"Route '{route.name}' output model '{output_model.name}' "
-                f"must not be an ingress model (is_ingress must be False)."
+                "must not be an ingress model (is_ingress must be False)."
             )
 
 
@@ -140,7 +140,7 @@ def _validate_models(raw_config: RawConfig) -> None:
         if model.is_ingress and model.schema_ is None:
             raise ConfigValidationError(
                 f"Model '{model.name}' is marked as ingress but does not have a schema defined. "
-                f"Ingress models must have a schema defined."
+                "Ingress models must have a schema defined."
             )
         if not model.is_ingress and model.schema_:
             raise ConfigValidationError(
@@ -156,7 +156,7 @@ def _validate_graph_and_add_order(raw_config: RawConfig) -> ValidatedConfig:
 
         This method performs the following steps:
         1. Validates that the directed graph formed by the models and routes is acyclic
-        and had unique path properties.
+        and has unique path properties.
         2. Computes a topological ordering of the models.
         3. Returns a new `ValidatedConfig` object where each model is wrapped as an
     `OrderedRawModel` with the corresponding `order` assigned.

src/ets/inject.py

@@ -51,8 +51,6 @@
 from ets.core.config_updater import ConfigUpdater
 from ets.core.model_derivation import ModelDeriver
 from ets.ports.inbound.aem_pack_registry import AEMPackRegistryPort
-from ets.ports.inbound.config_manager import ConfigManagerPort
-from ets.ports.inbound.config_updater import ConfigUpdaterPort
 from ets.ports.outbound.config_loader import ConfigLoaderPort
 from ets.ports.outbound.config_lock import ConfigLockPort
 from ets.ports.outbound.config_version import ConfigVersionerPort
@@ -65,7 +63,7 @@
 class _BaseWiring:
     """Contains everything reused across all higher-level preparation steps."""
 
-    config_manager: ConfigManagerPort
+    config_manager: ConfigManager
     config_lock: ConfigLockPort
     versioner: ConfigVersionerPort
     incoming_aem_pack_queue: IncomingAEMPackQueuePort
@@ -141,9 +139,7 @@ async def _prepare_base_wiring(
 
 
 @asynccontextmanager
-async def prepare_config_updater(
-    *, config: Config
-) -> AsyncGenerator[ConfigUpdaterPort]:
+async def prepare_config_updater(*, config: Config) -> AsyncGenerator[ConfigUpdater]:
     """Construct and initialize a ConfigUpdater with all its dependencies."""
     async with (
         ConfiguredMongoClient(config=config) as client,
@@ -153,7 +149,6 @@ async def prepare_config_updater(
             input_config_path=config.input_config_path,
             config_lock=base.config_lock,
             config_manager=base.config_manager,
-            versioner=base.versioner,
             incoming_aem_pack_queue=base.incoming_aem_pack_queue,
         )
 

src/ets/ports/inbound/config_manager.py

@@ -23,12 +23,11 @@
 from yaml import safe_load
 
 from ets.core import config_manager as config_manager_module
-from ets.core.config_manager import ConfigManager
+from ets.core.config_manager import ConfigManager, ConfigManagerError
 from ets.core.config_pruning import prune_unproductive_subgraphs
 from ets.core.config_validation import ConfigValidationError
 from ets.core.model_derivation import ModelDeriver
 from ets.core.models import Model, PersistedConfig, RawConfig, ValidatedConfig
-from ets.ports.inbound.config_manager import ConfigManagerError
 from ets.ports.outbound.config_loader import ConfigLoaderPort
 from ets.ports.outbound.config_version import ConfigVersionerPort
 from ets.ports.outbound.config_writer import ConfigWriterPort