Add globals feature for OmegaConfigLoader using a globals resolver

kedro/config/omegaconf_config.py

@@ -103,12 +103,14 @@ def __init__(  # noqa: too-many-arguments
         """
         self.base_env = base_env
         self.default_run_env = default_run_env
+        self.globals = {}
 
         self.config_patterns = {
             "catalog": ["catalog*", "catalog*/**", "**/catalog*"],
             "parameters": ["parameters*", "parameters*/**", "**/parameters*"],
             "credentials": ["credentials*", "credentials*/**", "**/credentials*"],
             "logging": ["logging*", "logging*/**", "**/logging*"],
+            "globals": ["globals*", "globals*/**", "**/globals*"],
         }
         self.config_patterns.update(config_patterns or {})
 
@@ -117,7 +119,7 @@ def __init__(  # noqa: too-many-arguments
         # Register user provided custom resolvers
         if custom_resolvers:
             self._register_new_resolvers(custom_resolvers)
-
+        self._register_globals_resolver()
         file_mimetype, _ = mimetypes.guess_type(conf_source)
         if file_mimetype == "application/x-tar":
             self._protocol = "tar"
@@ -136,10 +138,20 @@ def __init__(  # noqa: too-many-arguments
             env=env,
             runtime_params=runtime_params,
         )
+        # Read globals from patterns if the files exist
+        try:
+            self.globals = self._get_config("globals")
+        except MissingConfigException:
+            pass
+
+    def __repr__(self):  # pragma: no cover
+        return (
+            f"OmegaConfigLoader(conf_source={self.conf_source}, env={self.env}, "
+            f"config_patterns={self.config_patterns})"
+        )
 
     def __getitem__(self, key) -> dict[str, Any]:
-        """Get configuration files by key, load and merge them, and
-        return them in the form of a config dictionary.
+        """Get configuration files by key and return them in the form of a config dictionary.
 
         Args:
             key: Key of the configuration type to fetch.
@@ -158,32 +170,29 @@ def __getitem__(self, key) -> dict[str, Any]:
         # explicitly on the ``OmegaConfigLoader`` instance.
         if key in self:
             return super().__getitem__(key)
-
         if key not in self.config_patterns:
             raise KeyError(
                 f"No config patterns were found for '{key}' in your config loader"
             )
-        patterns = [*self.config_patterns[key]]
+        return self._get_config(key)
 
+    def _get_config(self, key) -> dict[str, Any]:
+        """Helper function to get configuration files by key, load and merge them,
+        and return them in the form of a config dictionary"""
+        patterns = [*self.config_patterns[key]]
         read_environment_variables = key == "credentials"
 
         processed_files: set[Path] = set()
         # Load base env config
-        if self._protocol == "file":
-            base_path = str(Path(self.conf_source) / self.base_env)
-        else:
-            base_path = str(Path(self._fs.ls("", detail=False)[-1]) / self.base_env)
+        base_path = self._get_conf_path(self.base_env)
         base_config = self.load_and_merge_dir_config(
             base_path, patterns, key, processed_files, read_environment_variables
         )
         config = base_config
 
         # Load chosen env config
         run_env = self.env or self.default_run_env
-        if self._protocol == "file":
-            env_path = str(Path(self.conf_source) / run_env)
-        else:
-            env_path = str(Path(self._fs.ls("", detail=False)[-1]) / run_env)
+        env_path = self._get_conf_path(run_env)
         env_config = self.load_and_merge_dir_config(
             env_path, patterns, key, processed_files, read_environment_variables
         )
@@ -199,18 +208,19 @@ def __getitem__(self, key) -> dict[str, Any]:
 
         config.update(env_config)
 
-        if not processed_files:
+        if not processed_files and key != "globals":
             raise MissingConfigException(
                 f"No files of YAML or JSON format found in {base_path} or {env_path} matching"
                 f" the glob pattern(s): {[*self.config_patterns[key]]}"
             )
         return config
 
-    def __repr__(self):  # pragma: no cover
-        return (
-            f"OmegaConfigLoader(conf_source={self.conf_source}, env={self.env}, "
-            f"config_patterns={self.config_patterns})"
-        )
+    def _get_conf_path(self, env):
+        if self._protocol == "file":
+            conf_path = str(Path(self.conf_source) / env)
+        else:
+            conf_path = str(Path(self._fs.ls("", detail=False)[-1]) / env)
+        return conf_path
 
     def load_and_merge_dir_config(  # noqa: too-many-arguments
         self,
@@ -246,36 +256,12 @@ def load_and_merge_dir_config(  # noqa: too-many-arguments
                 f"Given configuration path either does not exist "
                 f"or is not a valid directory: {conf_path}"
             )
-
-        paths = [
-            Path(each)
-            for pattern in patterns
-            for each in self._fs.glob(Path(f"{str(conf_path)}/{pattern}").as_posix())
-        ]
-        deduplicated_paths = set(paths)
-        config_files_filtered = [
-            path for path in deduplicated_paths if self._is_valid_config_path(path)
-        ]
-
-        config_per_file = {}
-        for config_filepath in config_files_filtered:
-            try:
-                with self._fs.open(str(config_filepath.as_posix())) as open_config:
-                    # As fsspec doesn't allow the file to be read as StringIO,
-                    # this is a workaround to read it as a binary file and decode it back to utf8.
-                    tmp_fo = io.StringIO(open_config.read().decode("utf8"))
-                    config = OmegaConf.load(tmp_fo)
-                    processed_files.add(config_filepath)
-                if read_environment_variables:
-                    self._resolve_environment_variables(config)
-                config_per_file[config_filepath] = config
-            except (ParserError, ScannerError) as exc:
-                line = exc.problem_mark.line  # type: ignore
-                cursor = exc.problem_mark.column  # type: ignore
-                raise ParserError(
-                    f"Invalid YAML or JSON file {Path(conf_path, config_filepath.name).as_posix()},"
-                    f" unable to read line {line}, position {cursor}."
-                ) from exc
+        config_files_filtered = self._get_filtered_conf_paths(conf_path, patterns)
+        config_per_file = self._load_config_files(
+            conf_path, config_files_filtered, read_environment_variables
+        )
+        for file in config_per_file.keys():
+            processed_files.add(file)
 
         seen_file_to_keys = {
             file: set(config.keys()) for file, config in config_per_file.items()
@@ -299,6 +285,18 @@ def load_and_merge_dir_config(  # noqa: too-many-arguments
             if not k.startswith("_")
         }
 
+    def _get_filtered_conf_paths(self, conf_path, patterns):
+        paths = [
+            Path(each)
+            for pattern in patterns
+            for each in self._fs.glob(Path(f"{str(conf_path)}/{pattern}").as_posix())
+        ]
+        deduplicated_paths = set(paths)
+        config_files_filtered = [
+            path for path in deduplicated_paths if self._is_valid_config_path(path)
+        ]
+        return config_files_filtered
+
     def _is_valid_config_path(self, path):
         """Check if given path is a file path and file type is yaml or json."""
         posix_path = path.as_posix()
@@ -308,6 +306,39 @@ def _is_valid_config_path(self, path):
             ".json",
         ]
 
+    def _load_config_files(
+        self, conf_path, config_filepaths, read_environment_variables
+    ):
+        config_per_file = {}
+        for config_filepath in config_filepaths:
+            try:
+                with self._fs.open(str(config_filepath.as_posix())) as open_config:
+                    # As fsspec doesn't allow the file to be read as StringIO,
+                    # this is a workaround to read it as a binary file and decode it back to utf8.
+                    tmp_fo = io.StringIO(open_config.read().decode("utf8"))
+                    config = OmegaConf.load(tmp_fo)
+                if read_environment_variables:
+                    self._resolve_environment_variables(config)
+                config_per_file[config_filepath] = config
+            except (ParserError, ScannerError) as exc:
+                line = exc.problem_mark.line  # type: ignore
+                cursor = exc.problem_mark.column  # type: ignore
+                raise ParserError(
+                    f"Invalid YAML or JSON file {Path(conf_path, config_filepath.name).as_posix()},"
+                    f" unable to read line {line}, position {cursor}."
+                ) from exc
+        return config_per_file
+
+    def _register_globals_resolver(self):
+        """Register the globals resolver"""
+        OmegaConf.register_new_resolver(
+            "globals", lambda x: self._get_globals_value(x), replace=True
+        )
+
+    def _get_globals_value(self, variable):
+        """Return the globals values to the resolver"""
+        return self.globals[variable]
+
     @staticmethod
     def _register_new_resolvers(resolvers: dict[str, Callable]):
         """Register custom resolvers"""

tests/config/test_omegaconf_config.py

@@ -671,3 +671,19 @@ def test_custom_resolvers(self, tmp_path):
         assert conf["parameters"]["model_options"]["param1"] == 7
         assert conf["parameters"]["model_options"]["param2"] == 3
         assert conf["parameters"]["model_options"]["param3"] == "my_env_variable"
+
+    def test_globals(self, tmp_path):
+        base_params = tmp_path / _BASE_ENV / "parameters.yml"
+        globals_params = tmp_path / _BASE_ENV / "globals.yml"
+        param_config = {
+            "param1": "${globals:hello}",
+        }
+        globals_config = {
+            "hello": 34,
+        }
+        _write_yaml(base_params, param_config)
+        _write_yaml(globals_params, globals_config)
+        conf = OmegaConfigLoader(tmp_path, default_run_env="")
+        parameters = conf["parameters"]
+        assert OmegaConf.has_resolver("globals")
+        assert parameters["param1"] == 34