diff --git a/cognite_toolkit/_cdf_tk/commands/build_v2/build_cmd.py b/cognite_toolkit/_cdf_tk/commands/build_v2/build_cmd.py index a18cbb2b83..2fe70efbd7 100644 --- a/cognite_toolkit/_cdf_tk/commands/build_v2/build_cmd.py +++ b/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 diff --git a/cognite_toolkit/_cdf_tk/commands/build_v2/build_input.py b/cognite_toolkit/_cdf_tk/commands/build_v2/build_input.py index 1412eff4c6..1d86dd27cc 100644 --- a/cognite_toolkit/_cdf_tk/commands/build_v2/build_input.py +++ b/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,24 +42,24 @@ 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(): @@ -66,20 +67,22 @@ def _load_config( 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), ) diff --git a/cognite_toolkit/_cdf_tk/data_classes/modules.py b/cognite_toolkit/_cdf_tk/data_classes/modules.py new file mode 100644 index 0000000000..db92c75386 --- /dev/null +++ b/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} diff --git a/tests/test_unit/test_cdf_tk/test_data_classes/test_modules.py b/tests/test_unit/test_cdf_tk/test_data_classes/test_modules.py new file mode 100644 index 0000000000..2bbe44faca --- /dev/null +++ b/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", + }