[CDF-26544] Pydantic modules

cognite_toolkit/_cdf_tk/commands/build_v2/build_cmd.py

@@ -1,6 +1,7 @@
 from pathlib import Path
 from typing import Any, Literal, TypedDict
 
+from _cdf_tk.data_classes.modules import ModulesDirectory
 from rich import print
 from rich.panel import Panel
 
@@ -121,7 +122,7 @@ def _validate_modules(self, input: BuildInput) -> BuildIssueList:
         # Validate module selection
         user_selected_modules = input.config.environment.get_selected_modules({})
         module_warnings = validate_module_selection(
-            modules=input.modules,
+            modules=ModulesDirectory.load(input.organization_dir, input.config.environment.selected),
             config=input.config,
             packages={},
             selected_modules=user_selected_modules,
@@ -146,7 +147,7 @@ def _validate_modules(self, input: BuildInput) -> BuildIssueList:
     def _build_configuration(self, input: BuildInput) -> tuple[BuiltModuleList, BuildIssueList]:
         issues = BuildIssueList()
         # Use input.modules.selected directly (it's already a ModuleDirectories)
-        if not input.modules.selected:
+        if not list(input.config.environment.selected):
             return BuiltModuleList(), issues
 
         # first collect variables into practical lookup

cognite_toolkit/_cdf_tk/commands/build_v2/build_input.py

@@ -2,6 +2,8 @@
 from functools import cached_property
 from pathlib import Path
 
+from cognite_toolkit._cdf_tk.data_classes.modules import ModulesDirectory
+
 if sys.version_info >= (3, 11):
     from typing import Self
 else:
@@ -14,7 +16,6 @@
 from cognite_toolkit._cdf_tk.data_classes import (
     BuildConfigYAML,
     BuildVariables,
-    ModuleDirectories,
 )
 from cognite_toolkit._cdf_tk.tk_warnings import ToolkitWarning, WarningList
 from cognite_toolkit._cdf_tk.utils.modules import parse_user_selected_modules
@@ -31,8 +32,8 @@ class BuildInput(BaseModel):
     build_env_name: str
     config: BuildConfigYAML
     client: ToolkitClient | None = None
-    selected: list[str | Path] | None = None
     warnings: WarningList[ToolkitWarning] | None = None
+    user_selected: list[str | Path] | None = None
 
     @classmethod
     def load(
@@ -41,45 +42,47 @@ def load(
         build_dir: Path,
         build_env_name: str | None,
         client: ToolkitClient | None,
-        selected: list[str | Path] | None = None,
+        user_selected: list[str | Path] | None = None,
     ) -> Self:
         resolved_org_dir = Path.cwd() if organization_dir in {Path("."), Path("./")} else organization_dir
         resolved_env = build_env_name or DEFAULT_ENV
-        config, warnings = cls._load_config(resolved_org_dir, resolved_env, selected)
+        config, warnings = cls._load_config(resolved_org_dir, resolved_env, user_selected)
         return cls(
             organization_dir=resolved_org_dir,
             build_dir=build_dir,
             build_env_name=resolved_env,
             config=config,
             client=client,
-            selected=selected,
             warnings=warnings,
+            user_selected=user_selected,
         )
 
     @classmethod
     def _load_config(
-        cls, organization_dir: Path, build_env_name: str, selected: list[str | Path] | None
+        cls, organization_dir: Path, build_env_name: str, user_selected: list[str | Path] | None
     ) -> tuple[BuildConfigYAML, WarningList[ToolkitWarning]]:
         warnings: WarningList[ToolkitWarning] = WarningList[ToolkitWarning]()
         if (organization_dir / BuildConfigYAML.get_filename(build_env_name or DEFAULT_ENV)).exists():
             config = BuildConfigYAML.load_from_directory(organization_dir, build_env_name or DEFAULT_ENV)
         else:
             # Loads the default environment
             config = BuildConfigYAML.load_default(organization_dir)
-        if selected:
-            config.environment.selected = parse_user_selected_modules(selected, organization_dir)
+        if user_selected:
+            config.environment.selected = list(set(parse_user_selected_modules(list(user_selected), organization_dir)))
         config.set_environment_variables()
         if environment_warning := config.validate_environment():
             warnings.append(environment_warning)
         return config, warnings
 
     @cached_property
-    def modules(self) -> ModuleDirectories:
-        user_selected_modules = self.config.environment.get_selected_modules({})
-        return ModuleDirectories.load(self.organization_dir, user_selected_modules)
+    def modules(self) -> ModulesDirectory:
+        selection = self.user_selected or self.config.environment.selected
+        return ModulesDirectory.load(self.organization_dir, selection)
 
     @cached_property
     def variables(self) -> BuildVariables:
         return BuildVariables.load_raw(
-            self.config.variables, self.modules.available_paths, self.modules.selected.available_paths
+            self.config.variables,
+            self.modules.available_paths,
+            set(Path(sel) for sel in self.config.environment.selected),
         )

cognite_toolkit/_cdf_tk/data_classes/modules.py

@@ -0,0 +1,81 @@
+import sys
+from functools import cached_property
+from pathlib import Path
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cognite_toolkit._cdf_tk.constants import MODULES
+from cognite_toolkit._cdf_tk.utils import iterate_modules
+from cognite_toolkit._cdf_tk.utils.modules import parse_user_selected_modules
+
+from ._module_toml import ModuleToml
+
+if sys.version_info >= (3, 11):
+    from typing import Self
+else:
+    from typing_extensions import Self
+
+
+class Resource(BaseModel):
+    model_config = ConfigDict(
+        frozen=True,
+        validate_assignment=True,
+    )
+
+    path: Path
+
+    @classmethod
+    def load(cls, path: Path) -> Self:
+        return cls(path=path)
+
+
+class Module(BaseModel):
+    model_config = ConfigDict(
+        frozen=True,
+        validate_assignment=True,
+    )
+
+    path: Path
+    resources: list[Resource]
+    definition: ModuleToml | None = None
+
+    @classmethod
+    def load(cls, path: Path, resource_paths: list[Path]) -> Self:
+        definition = ModuleToml.load(path / ModuleToml.filename) if (path / ModuleToml.filename).exists() else None
+        resources = [Resource.load(path=resource_path) for resource_path in resource_paths]
+        return cls(path=path, resources=resources, definition=definition)
+
+
+class ModulesDirectory(BaseModel):
+    model_config = ConfigDict(
+        frozen=True,
+        validate_assignment=True,
+    )
+
+    modules: list[Module] = Field(default_factory=list)
+
+    @classmethod
+    def load(cls, organization_dir: Path, selection: list[str | Path] | None = None) -> Self:
+        selected = parse_user_selected_modules(selection, organization_dir) if selection else None
+        return cls(
+            modules=[
+                Module.load(path=module_path, resource_paths=resource_paths)
+                for module_path, resource_paths in iterate_modules(organization_dir / MODULES)
+                if cls._is_selected(module_path, organization_dir, selected)
+            ],
+        )
+
+    @staticmethod
+    def _is_selected(module_path: Path, organization_dir: Path, selection: list[str | Path] | None) -> bool:
+        if selection is None:
+            return True
+        relative = module_path.relative_to(organization_dir)
+        return module_path.name in selection or relative in selection or any(p in selection for p in relative.parents)
+
+    @cached_property
+    def paths(self) -> list[Path]:
+        return [module.path for module in self.modules]
+
+    @cached_property
+    def available_paths(self) -> set[Path]:
+        return {module.path for module in self.modules}

tests/test_unit/test_cdf_tk/test_data_classes/test_modules.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+from cognite_toolkit._cdf_tk.constants import MODULES
+from cognite_toolkit._cdf_tk.data_classes.modules import ModulesDirectory
+from tests.data import COMPLETE_ORG
+
+
+class TestModules:
+    def test_load_modules(self) -> None:
+        modules = ModulesDirectory.load(COMPLETE_ORG)
+
+        assert len(modules.modules) == 3
+        assert {module.path for module in modules.modules} == {
+            COMPLETE_ORG / MODULES / "my_example_module",
+            COMPLETE_ORG / MODULES / "my_file_expand_module",
+            COMPLETE_ORG / MODULES / "populate_model",
+        }
+
+    def test_load_selection(self) -> None:
+        modules = ModulesDirectory.load(
+            COMPLETE_ORG, selection=["my_example_module", Path(MODULES) / "my_file_expand_module"]
+        )
+
+        assert len(modules.modules) == 2
+        assert {module.path for module in modules.modules} == {
+            COMPLETE_ORG / MODULES / "my_example_module",
+            COMPLETE_ORG / MODULES / "my_file_expand_module",
+        }