Source code for kedro.config.config

"""This module provides ``kedro.config`` with the functionality to load one
or more configuration files from specified paths.
from __future__ import annotations

from pathlib import Path
from typing import Any, Iterable

from kedro.config.abstract_config import AbstractConfigLoader
from kedro.config.common import _get_config_from_patterns, _remove_duplicates

[docs]class ConfigLoader(AbstractConfigLoader): """Recursively scan directories (config paths) contained in ``conf_source`` for configuration files with a ``yaml``, ``yml``, ``json``, ``ini``, ``pickle``, ``xml`` or ``properties`` extension, load them, and return them in the form of a config dictionary. The first processed config path is the ``base`` directory inside ``conf_source``. The optional ``env`` argument can be used to specify a subdirectory of ``conf_source`` to process as a config path after ``base``. When the same top-level key appears in any 2 config files located in the same (sub)directory, a ``ValueError`` is raised. When the same key appears in any 2 config files located in different (sub)directories, the last processed config path takes precedence and overrides this key. For example, if your ``conf_source`` looks like this: :: . `-- conf |-- |-- base | |-- catalog.yml | |-- logging.yml | `-- experiment1 | `-- parameters.yml `-- local |-- catalog.yml |-- db.ini |-- experiment1 | |-- parameters.yml | `-- model_parameters.yml `-- experiment2 `-- parameters.yml You can access the different configurations as follows: :: >>> import logging.config >>> from kedro.config import ConfigLoader >>> from kedro.framework.project import settings >>> >>> conf_path = str(project_path / settings.CONF_SOURCE) >>> conf_loader = ConfigLoader(conf_source=conf_path, env="local") >>> >>> conf_logging = conf_loader["logging"] >>> logging.config.dictConfig(conf_logging) # set logging conf >>> >>> conf_catalog = conf_loader["catalog"] >>> conf_params = conf_loader["parameters"] """
[docs] def __init__( # noqa: PLR0913 self, conf_source: str, env: str = None, runtime_params: dict[str, Any] = None, config_patterns: dict[str, list[str]] = None, *, base_env: str = "base", default_run_env: str = "local", ): """Instantiates a ``ConfigLoader``. Args: conf_source: Path to use as root directory for loading configuration. env: Environment that will take precedence over base. runtime_params: Extra parameters passed to a Kedro run. config_patterns: Regex patterns that specify the naming convention for configuration files so they can be loaded. Can be customised by supplying config_patterns as in `CONFIG_LOADER_ARGS` in ``. base_env: Name of the base environment. Defaults to `"base"`. This is used in the `conf_paths` property method to construct the configuration paths. default_run_env: Name of the default run environment. Defaults to `"local"`. This is used in the `conf_paths` property method to construct the configuration paths. Can be overriden by supplying the `env` argument. """ self.base_env = base_env self.default_run_env = default_run_env self.config_patterns = { "catalog": ["catalog*", "catalog*/**", "**/catalog*"], "parameters": ["parameters*", "parameters*/**", "**/parameters*"], "credentials": ["credentials*", "credentials*/**", "**/credentials*"], "logging": ["logging*", "logging*/**", "**/logging*"], } self.config_patterns.update(config_patterns or {}) super().__init__( conf_source=conf_source, env=env, runtime_params=runtime_params, )
def __getitem__(self, key): # Allow bypassing of loading config from patterns if a key and value have been set # explicitly on the ``ConfigLoader`` instance. if key in self: return super().__getitem__(key) return self.get(*self.config_patterns[key]) def __repr__(self): # pragma: no cover return ( f"ConfigLoader(conf_source={self.conf_source}, env={self.env}, " f"config_patterns={self.config_patterns})" ) @property def conf_paths(self): """Property method to return deduplicated configuration paths.""" return _remove_duplicates(self._build_conf_paths())
[docs] def get(self, *patterns: str) -> dict[str, Any]: # type: ignore return _get_config_from_patterns( conf_paths=self.conf_paths, patterns=list(patterns) )
def _build_conf_paths(self) -> Iterable[str]: run_env = self.env or self.default_run_env return [ str(Path(self.conf_source) / self.base_env), str(Path(self.conf_source) / run_env), ]