kedro.config.OmegaConfigLoader

class kedro.config.OmegaConfigLoader(conf_source, env=None, runtime_params=None, *, config_patterns=None, base_env=None, default_run_env=None, custom_resolvers=None, merge_strategy=None)[source]

Recursively scan directories (config paths) contained in conf_source for configuration files with a yaml, yml or json extension, load and merge them through OmegaConf (https://omegaconf.readthedocs.io/) 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 two config files located in the same (sub)directory, a ValueError is raised.

When the same key appears in any two config files located in different (sub)directories, the last processed config path takes precedence and overrides this key and any sub-keys.

You can access the different configurations as follows:

import logging.config
from kedro.config import OmegaConfigLoader
from kedro.framework.project import settings

conf_path = str(project_path / settings.CONF_SOURCE)
conf_loader = OmegaConfigLoader(conf_source=conf_path, env="local")

conf_catalog = conf_loader["catalog"]
conf_params = conf_loader["parameters"]

OmegaConf supports variable interpolation in configuration https://omegaconf.readthedocs.io/en/2.2_branch/usage.html#merging-configurations. It is recommended to use this instead of yaml anchors with the OmegaConfigLoader.

This version of the OmegaConfigLoader does not support any of the built-in OmegaConf resolvers. Support for resolvers might be added in future versions.

To use this class, change the setting for the CONFIG_LOADER_CLASS constant in settings.py.

Example:

# in settings.py
from kedro.config import OmegaConfigLoader

CONFIG_LOADER_CLASS = OmegaConfigLoader

Methods

clear()

copy()

fromkeys(iterable[, value])

get(key[, default])

rtype:

Any

items()

keys()

rtype:

KeysView

load_and_merge_dir_config(conf_path, ...[, ...])

Recursively load and merge all configuration files in a directory using OmegaConf, which satisfy a given list of glob patterns from a specific path.

pop(k[,d])

If key is not found, d is returned if given, otherwise KeyError is raised.

popitem()

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[,d])

update([E, ]**F)

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

values()

clear() → None.  Remove all items from D.[source]
copy()[source]
classmethod fromkeys(iterable, value=None)[source]
get(key, default=None)[source]
Return type:

Any

items() → a set-like object providing a view on D's items[source]
keys()[source]
Return type:

KeysView

load_and_merge_dir_config(conf_path, patterns, key, processed_files, read_environment_variables=False)[source]

Recursively load and merge all configuration files in a directory using OmegaConf, which satisfy a given list of glob patterns from a specific path.

Parameters:

  • conf_path – Path to configuration directory.

  • patterns – List of glob patterns to match the filenames against.

  • key – Key of the configuration type to fetch.

  • processed_files – Set of files read for a given configuration type.

  • read_environment_variables – Whether to resolve environment variables.

Raises:

  • MissingConfigException – If configuration path doesn’t exist or isn’t valid.

  • ValueError – If two or more configuration files contain the same key(s).

  • ParserError – If config file contains invalid YAML or JSON syntax.

Returns:

Resulting configuration dictionary.

pop(k[, d]) → v, remove specified key and return the corresponding value.[source]

If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), remove and return some (key, value) pair[source]

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D[source]
update([E, ]**F) → None.  Update D from mapping/iterable E and F.[source]

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

values() → an object providing a view on D's values[source]