Add Toolkit template variable substitution for toolkit YAML reads

cognite/neat/_data_model/importers/_api_importer.py

@@ -9,6 +9,12 @@
 from cognite.neat._data_model._analysis import ValidationResources
 from cognite.neat._data_model._snapshot import SchemaSnapshot
 from cognite.neat._data_model.importers._base import DMSImporter
+from cognite.neat._data_model.importers._toolkit_variables import (
+    ToolkitProjectContext,
+    ToolkitReadOptions,
+    _is_filesystem_path,
+    prepare_toolkit_yaml_content,
+)
 from cognite.neat._data_model.models.dms import (
     DataModelReference,
     RequestSchema,
@@ -118,18 +124,29 @@ def from_cdf(cls, data_model: DataModelReference, client: NeatClient) -> "DMSAPI
         )
 
     @classmethod
-    def from_yaml(cls, yaml_file: Path, data_model_file: Path | None = None) -> "DMSAPIImporter":
+    def from_yaml(
+        cls,
+        yaml_file: Path,
+        data_model_file: Path | str | None = None,
+        *,
+        toolkit_env: str | None = None,
+        toolkit_config: Path | str | None = None,
+        toolkit_version: str | None = None,
+        substitute_toolkit_variables: bool = True,
+        options: ToolkitReadOptions | None = None,
+    ) -> "DMSAPIImporter":
         """Create a DMSTableImporter from a YAML file."""
+        options = options or ToolkitReadOptions.from_args(
+            toolkit_env,
+            toolkit_config,
+            toolkit_version,
+            substitute_toolkit_variables=substitute_toolkit_variables,
+        )
         source = cls._display_name(yaml_file)
         if yaml_file.suffix.lower() in {".yaml", ".yml", ".json"}:
-            yaml_content = yaml_file.read_text(encoding=cls.ENCODING)
-            # DataModels and Views have `version` that is often given as an integer by the user.
-            # This ensures that the version is always read as a string, even if the user forgets to
-            # quote it in the YAML file.
-            fixed_content = quote_int_value_by_key_in_yaml(yaml_content, "version")
-            return cls(yaml.safe_load(fixed_content))
+            return cls(cls._load_toolkit_yaml_file(yaml_file, options))
         elif yaml_file.is_dir():
-            return cls(cls._read_yaml_files(yaml_file, data_model_file))
+            return cls(cls._read_yaml_files(yaml_file, data_model_file, options=options))
         raise FileReadException(source.as_posix(), f"Unsupported file type: {source.suffix}")
 
     @classmethod
@@ -147,27 +164,103 @@ def _display_name(cls, filepath: Path) -> Path:
         return source
 
     @classmethod
-    def _read_yaml_files(cls, directory: Path, data_model_file: Path | None = None) -> dict[str, Any]:
-        """Read all YAML files in a directory and combine them into a single dictionary."""
+    def _load_toolkit_yaml_file(
+        cls,
+        yaml_file: Path,
+        options: ToolkitReadOptions,
+        *,
+        variables: dict[str, str] | None = None,
+        quote_version: bool = True,
+        data_model_dir: Path | None = None,
+    ) -> Any:
+        yaml_content = yaml_file.read_text(encoding=cls.ENCODING)
+        yaml_content = prepare_toolkit_yaml_content(
+            yaml_content,
+            source=yaml_file,
+            data_model_dir=data_model_dir or yaml_file.parent,
+            variables=variables,
+            options=options,
+        )
+        if quote_version:
+            # DataModels and Views often give `version` as an unquoted integer.
+            yaml_content = quote_int_value_by_key_in_yaml(yaml_content, "version")
+        return yaml.safe_load(yaml_content)
+
+    @classmethod
+    def _is_toolkit_schema_yaml(cls, yaml_file: Path) -> bool:
+        """True for DMS resource files; skip Toolkit config, Yamale schemas, and other YAML.
+
+        Resource files follow Toolkit naming (``*.Container.yaml``, ``*.View.yaml``, …).
+        Names such as ``data_model_container.yaml`` are validation schemas, not DMS resources.
+        """
+        if yaml_file.suffix.lower() not in {".yaml", ".yml", ".json"}:
+            return False
+        name = yaml_file.name.casefold()
+        return any(
+            name.endswith(suffix)
+            for kind in ("datamodel", "view", "container", "space", "node")
+            for suffix in (f".{kind}.yaml", f".{kind}.yml", f".{kind}.json")
+        )
+
+    @classmethod
+    def _read_yaml_files(
+        cls,
+        directory: Path,
+        data_model_file: Path | str | None = None,
+        *,
+        options: ToolkitReadOptions | None = None,
+    ) -> dict[str, Any]:
+        """Read Toolkit DMS YAML files in a directory and combine them into a single schema.
+
+        Template variables are resolved per file from that file's module path, so a read of
+        ``modules/`` still sees nested keys such as ``variables.modules.valve_track.*``.
+        When ``data_model_file`` selects one model, sibling DataModel/view/node files are
+        not parsed.
+        """
+        options = options or ToolkitReadOptions()
         schema_data: dict[str, Any] = {}
         data_model: dict[str, Any] | None = None
-        for yaml_file in directory.rglob("**/*"):
-            if yaml_file.suffix.lower() not in {".yaml", ".yml", ".json"}:
-                continue
+        dm_path = Path(data_model_file) if data_model_file is not None else None
+        data_model_file_name = dm_path.name if dm_path is not None else None
+        config_dir = directory
+        if dm_path is not None and _is_filesystem_path(dm_path) and dm_path.is_file():
+            config_dir = dm_path.parent
+        context = ToolkitProjectContext.load(config_dir, options) if options.substitute else None
+
+        schema_files = [yaml_file for yaml_file in directory.rglob("**/*") if cls._is_toolkit_schema_yaml(yaml_file)]
+        selected_dm_files = [
+            yaml_file
+            for yaml_file in schema_files
+            if yaml_file.stem.casefold().endswith("datamodel")
+            and (not data_model_file_name or yaml_file.name == data_model_file_name)
+        ]
+        resource_root: Path | None = None
+        if data_model_file_name and len(selected_dm_files) == 1 and _is_filesystem_path(selected_dm_files[0]):
+            resource_root = selected_dm_files[0].parent
+
+        for yaml_file in schema_files:
             stem = yaml_file.stem.casefold()
-
-            yaml_content = yaml_file.read_text(encoding=cls.ENCODING)
-            if stem.endswith("datamodel") or stem.endswith("view"):
-                # DataModels and Views have `version` that is often given as an integer by the user.
-                # This ensures that the version is always read as a string, even if the user forgets to
-                # quote it in the YAML file.
-                yaml_content = quote_int_value_by_key_in_yaml(yaml_content, "version")
-            data = yaml.safe_load(yaml_content)
+            if stem.endswith("datamodel") and data_model_file_name and yaml_file.name != data_model_file_name:
+                continue
+            if (
+                resource_root is not None
+                and stem.endswith(("view", "node"))
+                and not cls._is_path_under(yaml_file, resource_root)
+            ):
+                continue
+            quote_version = stem.endswith("datamodel") or stem.endswith("view")
+            file_dir = yaml_file.parent
+            file_variables = context.variables_for(file_dir, options.version) if context is not None else None
+            data = cls._load_toolkit_yaml_file(
+                yaml_file,
+                options,
+                variables=file_variables,
+                quote_version=quote_version,
+                data_model_dir=file_dir,
+            )
             list_data = data if isinstance(data, list) else [data]
 
             if stem.endswith("datamodel"):
-                if data_model_file and yaml_file.name != data_model_file.name:
-                    continue  # skip this file as it doesn't match the specified data model file
                 if data_model is not None and not data_model_file:
                     raise FileReadException(
                         cls._display_name(directory).as_posix(),
@@ -194,6 +287,14 @@ def _read_yaml_files(cls, directory: Path, data_model_file: Path | None = None)
         schema_data["dataModel"] = data_model
         return schema_data
 
+    @staticmethod
+    def _is_path_under(path: Path, root: Path) -> bool:
+        try:
+            path.resolve().relative_to(root.resolve())
+        except (ValueError, OSError):
+            return False
+        return True
+
 
 class DMSAPICreator(DMSImporter):
     """Creates data model from the existing schema components in CDF"""