Configuration API Reference

from kedro.config import OmegaConfigLoader, AbstractConfigLoader
from kedro.config.omegaconf_config import MergeStrategies

from collections import UserDict

class AbstractConfigLoader(UserDict):
    """
    Abstract base class for all ConfigLoader implementations.
    Extends UserDict to provide dict-like interface.

    Inherits all UserDict methods including:
    - keys() -> KeysView
    - values() -> ValuesView
    - items() -> ItemsView
    - pop(key, default=None) -> Any
    - update(other) -> None
    - clear() -> None
    """

    def __init__(
        self,
        conf_source: str | Path,
        env: str | None = None,
        runtime_params: dict[str, Any] | None = None,
        **kwargs: Any
    ):
        """
        Initialize the config loader.

        Parameters:
        - conf_source: Path to configuration directory
        - env: Environment name (e.g., 'local', 'prod'). If None, only base config is loaded
        - runtime_params: Dictionary of runtime parameter overrides
        - **kwargs: Additional keyword arguments for subclass customization

        Attributes set:
        - self.conf_source: Configuration source directory path
        - self.env: Current environment name or None
        - self.runtime_params: Runtime parameter overrides (empty dict if None provided)
        """

    def __getitem__(self, key: str) -> dict[str, Any]:
        """
        Get configuration by key.

        Parameters:
        - key: Configuration key

        Returns:
        Configuration dictionary
        """

    def __setitem__(self, key: str, value: dict[str, Any]) -> None:
        """
        Set configuration for a key.

        Parameters:
        - key: Configuration key
        - value: Configuration dictionary
        """

    def __contains__(self, key: str) -> bool:
        """
        Check if configuration key exists.

        Parameters:
        - key: Configuration key

        Returns:
        True if key exists
        """

    def __iter__(self) -> Iterator[str]:
        """
        Iterate over configuration keys.

        Returns:
        Iterator of configuration keys
        """

class OmegaConfigLoader:
    """Load and merge YAML/JSON configuration files using OmegaConf."""

    def __init__(
        self,
        conf_source: str,
        env: str | None = None,
        runtime_params: dict[str, Any] | None = None,
        config_patterns: dict[str, list[str]] | None = None,
        base_env: str | None = None,
        default_run_env: str | None = None,
        merge_strategy: dict[str, str] | None = None,
        custom_resolvers: dict[str, Callable] | None = None,
        ignore_hidden: bool = True
    ):
        """
        Initialize OmegaConfigLoader.

        Parameters:
        - conf_source: Directory path containing configuration files
        - env: Environment name (e.g., 'local', 'prod'), takes precedence over default_run_env
        - runtime_params: Parameters to override at runtime
        - config_patterns: Mapping of config keys to glob patterns for file matching.
          Allows custom configuration file patterns per config key.
          Default patterns:
            - "catalog": ["catalog*", "catalog/**"]
            - "parameters": ["parameters*", "parameters/**"]
            - "credentials": ["credentials*", "credentials/**"]
            - "logging": ["logging*", "logging/**"]
          Pattern matching:
            - Patterns are evaluated as globs relative to conf_source/env/
            - Multiple patterns per key are supported (matched in order)
            - Supports nested paths with "**" wildcard
            - Files are loaded in alphanumeric order within matched patterns
          Example: {"catalog": ["catalog*.yml", "catalog/**/*.yml"]}
        - base_env: Base environment directory name (default: "base")
        - default_run_env: Default environment when env not specified (default: "local")
        - merge_strategy: Merge strategy per config key (MergeStrategies.SOFT or MergeStrategies.DESTRUCTIVE).
          Defaults to DESTRUCTIVE (environment overrides base completely).
          SOFT performs recursive merge (environment extends base).
        - custom_resolvers: Custom OmegaConf resolvers for variable interpolation
        - ignore_hidden: Whether to ignore hidden files (default: True)
        """

    def __getitem__(self, key: str) -> dict[str, Any]:
        """Get configuration by key."""

    def __setitem__(self, key: str, value: dict[str, Any]) -> None:
        """Set configuration for a key (programmatically override config)."""

    def keys(self) -> list[str]:
        """Get all configuration keys."""

    def get(self, key: str, default: Any = None) -> Any:
        """Get configuration with default fallback."""

    def __repr__() -> str:
        """String representation showing config source and environment."""

from enum import Enum, auto

class MergeStrategies(Enum):
    """Configuration merge strategies."""
    SOFT = auto()         # Recursive merge
    DESTRUCTIVE = auto()  # Complete replacement (default)

class BadConfigException(Exception):
    """Raised when a configuration file cannot be loaded."""

class MissingConfigException(Exception):
    """Raised when no configuration files found."""

from kedro.config import OmegaConfigLoader

loader = OmegaConfigLoader(conf_source="conf", env="local")

# Load configurations
catalog = loader["catalog"]
parameters = loader["parameters"]

loader = OmegaConfigLoader(
    conf_source="conf",
    env="prod",
    runtime_params={
        "model.learning_rate": 0.01,
        "data.batch_size": 128
    }
)

loader = OmegaConfigLoader(
    conf_source="conf",
    merge_strategy={
        "catalog": "soft",          # Merge catalog configs
        "parameters": "destructive"  # Replace parameters completely
    }
)

conf/
├── base/
│   ├── catalog/
│   │   ├── raw.yml
│   │   ├── processed.yml
│   │   └── models.yml
│   ├── parameters/
│   │   ├── data_science.yml
│   │   └── data_engineering.yml
│   └── spark.yml
└── prod/
    ├── catalog/
    │   └── raw.yml  # Override raw data location
    └── spark.yml    # Override Spark configs

from kedro.config import OmegaConfigLoader

loader = OmegaConfigLoader(
    conf_source="conf",
    env="prod",
    config_patterns={
        "catalog": ["catalog*.yml", "catalog/**/*.yml"],     # Match catalog/*.yml
        "parameters": ["parameters*.yml", "parameters/**/*.yml"],
        "spark": ["spark*.yml"]                              # Match spark.yml
    }
)

# Loads in this order:
# 1. conf/base/catalog/*.yml (all files alphabetically)
# 2. conf/prod/catalog/*.yml (overrides matching datasets)
catalog = loader["catalog"]

# 1. conf/base/parameters/**/*.yml (all subdirectory files)
# 2. conf/prod/parameters/**/*.yml (if present)
params = loader["parameters"]

# 1. conf/base/spark.yml
# 2. conf/prod/spark.yml (merges or replaces based on merge_strategy)
spark_conf = loader["spark"]