OmegaConfigLoader

def __getitem__(self, key: str) -> dict[str, Any]:
    """Get configuration files by key, load and merge them, and
    return them in the form of a config dictionary.

    Args:
        key: Key of the configuration type to fetch.

    Raises:
        KeyError: If key provided isn't present in the config_patterns of this
            ``OmegaConfigLoader`` instance.
        MissingConfigException: If no configuration files exist matching the patterns
            mapped to the provided key.

    Returns:
        Dict[str, Any]:  A Python dictionary with the combined
            configuration from all configuration files.
    """
    # Allow bypassing of loading config from patterns if a key and value have been set
    # explicitly on the ``OmegaConfigLoader`` instance.

    # Re-register runtime params resolver in case it was previously deactivated
    self._register_runtime_params_resolver()

    if key in self:
        return super().__getitem__(key)  # type: ignore[no-any-return]

    if key not in self.config_patterns:
        raise KeyError(
            f"No config patterns were found for '{key}' in your config loader"
        )

    base_configs, env_configs, base_path, env_path, processed_files = (
        self._read_raw_config(key)
    )
    return self._resolve_from_raw_config(
        key, base_configs, env_configs, base_path, env_path, processed_files
    )

def __repr__(self) -> str:  # pragma: no cover
    return (
        f"OmegaConfigLoader(conf_source={self.conf_source}, env={self.env}, "
        f"runtime_params={self.runtime_params}, "
        f"config_patterns={self.config_patterns}, "
        f"base_env={self.base_env}), "
        f"default_run_env={self.default_run_env}), "
        f"custom_resolvers={self._custom_resolvers}), "
        f"merge_strategy={self.merge_strategy})"
    )

def __setitem__(self, key: str, value: Any) -> None:
    if key == "globals":
        # Update the cached value at self._globals since it is used by the globals resolver
        self._globals = value
    super().__setitem__(key, value)

def _check_duplicates(self, key: str, config_per_file: dict[Path, Any]) -> None:
    if key == "parameters":
        seen_files_to_keys = {
            file: self._get_all_keys(OmegaConf.to_container(config, resolve=False))
            for file, config in config_per_file.items()
        }
    else:
        seen_files_to_keys = {
            file: set(config.keys()) for file, config in config_per_file.items()
        }

    duplicates = []

    filepaths = list(seen_files_to_keys.keys())
    for i, filepath1 in enumerate(filepaths, 1):
        config1 = seen_files_to_keys[filepath1]
        for filepath2 in filepaths[i:]:
            config2 = seen_files_to_keys[filepath2]

            combined_keys = config1 & config2
            overlapping_keys = {
                key for key in combined_keys if not key.startswith("_")
            }

            if overlapping_keys:
                sorted_keys = ", ".join(sorted(overlapping_keys))
                if len(sorted_keys) > 100:  # noqa: PLR2004
                    sorted_keys = sorted_keys[:100] + "..."
                duplicates.append(
                    f"Duplicate keys found in {filepath1} and {filepath2}: {sorted_keys}"
                )

    if duplicates:
        dup_str = "\n".join(duplicates)
        raise ValueError(f"{dup_str}")

@staticmethod
def _destructive_merge(
    config: dict[str, Any], env_config: dict[str, Any], env_path: str
) -> dict[str, Any]:
    # Destructively merge the two env dirs. The chosen env will override base.
    common_keys = config.keys() & env_config.keys()
    if common_keys:
        sorted_keys = ", ".join(sorted(common_keys))
        msg = (
            "Config from path '%s' will override the following "
            "existing top-level config keys: %s"
        )
        _config_logger.debug(msg, env_path, sorted_keys)

    config.update(env_config)
    return config

def _get_all_keys(self, cfg: Any, parent_key: str = "") -> set[str]:
    keys: set[str] = set()

    for key, value in cfg.items():
        full_key = f"{parent_key}.{key}" if parent_key else key
        if isinstance(value, dict):
            keys.update(self._get_all_keys(value, full_key))
        else:
            keys.add(full_key)
    return keys

def _get_conf_env_path(self, env: str) -> str:
    """Get the full path to the configuration environment directory."""
    if self._protocol in CLOUD_PROTOCOLS or self._protocol in HTTP_PROTOCOLS:
        return f"{self._remote_root_path}/{env}"
    if self._protocol == "file":
        return str(Path(self.conf_source) / env)
    return str(Path(self._fs.ls("", detail=False)[-1]) / env)

def _get_globals_value(self, variable: str, default_value: Any = _NO_VALUE) -> Any:
    """Return the globals values to the resolver"""
    if variable.startswith("_"):
        raise InterpolationResolutionError(
            "Keys starting with '_' are not supported for globals."
        )

    if not self._globals_oc:
        self._globals_oc = OmegaConf.create(self._globals)

    interpolated_value = OmegaConf.select(
        self._globals_oc, variable, default=default_value
    )
    if interpolated_value != _NO_VALUE:
        return interpolated_value
    else:
        raise InterpolationResolutionError(
            f"Globals key '{variable}' not found and no default value provided."
        )

def _get_runtime_value(self, variable: str, default_value: Any = _NO_VALUE) -> Any:
    """Return the runtime params values to the resolver"""
    if not self._runtime_params_oc:
        self._runtime_params_oc = OmegaConf.create(self.runtime_params)

    # Resolved separately from the `default=` lookup below so we can tell
    # a real runtime_params value apart from a resolver default/fallback
    # (e.g. `${runtime_params:type, ${globals:...}}`) -- only the former
    # is attacker-influenced when runtime_params comes from an untrusted
    # caller, and only the former is tracked in `_runtime_params_hits`.
    found_value = OmegaConf.select(
        self._runtime_params_oc, variable, default=_NO_VALUE
    )
    if found_value is not _NO_VALUE:
        self._runtime_params_hits.add(variable)
        return found_value

    if default_value is not _NO_VALUE:
        return default_value

    raise InterpolationResolutionError(
        f"Runtime parameter '{variable}' not found and no default value provided."
    )

@typing.no_type_check
def _guard_runtime_params_in_catalog_type(
    self,
    aggregate_config: Iterable[DictConfig],
    merged_config_container: dict[str, Any],
) -> None:
    """Reject a dataset ``type`` that resolves through ``runtime_params:``,
    when ``self.restrict_runtime_params_type_selection`` is set.

    ``type`` selects the class Kedro dynamically imports and instantiates,
    so letting untrusted runtime params choose it allows arbitrary
    importable ``AbstractDataset`` subclasses, including ones with side
    effects on load (e.g. ``kedro_datasets.pickle.PickleDataset``).
    Checked at every nesting depth to also cover wrapper datasets such as
    ``CachedDataset`` and ``PartitionedDataset``.
    """
    if not self.restrict_runtime_params_type_selection:
        return

    raw_container = OmegaConf.to_container(
        OmegaConf.unsafe_merge(*aggregate_config), resolve=False
    )

    for dataset_name, raw_entry in raw_container.items():
        resolved_entry = merged_config_container.get(dataset_name)
        for raw_node, resolved_node in self._iter_nested_dicts(
            raw_entry, resolved_entry
        ):
            raw_type = raw_node.get("type")
            if (
                not isinstance(raw_type, str)
                or _RUNTIME_PARAMS_MARKER not in raw_type
            ):
                continue

            # A `type` string can reference `runtime_params:` more than
            # once (e.g. concatenated resolver calls); treat it as
            # attacker-influenced if *any* of them actually hit a
            # runtime_params value.
            variables = [
                v.strip() for v in _RUNTIME_PARAMS_VARIABLE_RE.findall(raw_type)
            ]
            if not variables or not any(
                v in self._runtime_params_hits for v in variables
            ):
                # `runtime_params:` appears in the raw string, but no
                # actual runtime_params value was used for it -- it fell
                # back to a resolver default (e.g. a `globals:` value), so
                # it isn't attacker-influenced.
                continue

            resolved_type = resolved_node.get("type")
            raise InterpolationResolutionError(
                f"Dataset '{dataset_name}' resolves a 'type' field via "
                f"the 'runtime_params:' resolver to '{resolved_type}', "
                "which is not permitted. Runtime parameters must not "
                "select which dataset class is instantiated, at any "
                "nesting depth."
            )

@staticmethod
def _initialise_filesystem_and_protocol(
    conf_source: str,
) -> tuple[fsspec.AbstractFileSystem, str]:
    """Set up the file system based on the file type or protocol detected in conf_source."""
    # Force string for regex
    conf_source = str(conf_source)
    # Check if it's an archive file
    file_mimetype, _ = mimetypes.guess_type(conf_source)
    if file_mimetype == "application/x-tar":
        return fsspec.filesystem(protocol="tar", fo=conf_source), "tar"
    elif file_mimetype in (
        "application/zip",
        "application/x-zip-compressed",
        "application/zip-compressed",
    ):
        return fsspec.filesystem(protocol="zip", fo=conf_source), "zip"

    # Parse to check for protocol
    options = _parse_filepath(conf_source)
    protocol = options["protocol"]

    # Create and return the appropriate filesystem
    if protocol in HTTP_PROTOCOLS or protocol in CLOUD_PROTOCOLS:
        # For HTTP and cloud storage protocols, create the appropriate filesystem
        return fsspec.filesystem(protocol=protocol), protocol
    else:
        # Default to local filesystem
        return fsspec.filesystem(protocol="file", fo=conf_source), "file"

def _is_hidden(self, path_str: str) -> bool:
    """Check if path contains any hidden directory or is a hidden file"""
    path = Path(path_str)
    conf_path = Path(self.conf_source).resolve().as_posix()
    if self._protocol == "file":
        path = path.resolve()
    posix_path = path.as_posix()
    if posix_path.startswith(conf_path):
        posix_path = posix_path.replace(conf_path, "")
    parts = posix_path.split(self._fs.sep)  # filesystem specific separator
    HIDDEN = "."
    # Check if any component (folder or file) starts with a dot (.)
    return any(part.startswith(HIDDEN) for part in parts)

def _is_valid_config_path(self, path: Path) -> bool:
    """Check if given path is a file path and file type is yaml or json."""
    posix_path = path.as_posix()
    return self._fs.isfile(str(posix_path)) and path.suffix in [
        ".yml",
        ".yaml",
        ".json",
    ]

@staticmethod
def _iter_nested_dicts(
    raw_node: Any, resolved_node: Any
) -> typing.Generator[tuple[dict, dict], None, None]:
    """Yield ``(raw_node, resolved_node)`` for ``raw_node`` and every dict
    nested within it, pairing each with the corresponding resolved
    sub-tree by key so callers can compare a field's raw (unresolved) and
    resolved values at any depth.
    """
    if not isinstance(raw_node, dict):
        return
    resolved_node = resolved_node if isinstance(resolved_node, dict) else {}
    yield raw_node, resolved_node
    for key, raw_value in raw_node.items():
        if isinstance(raw_value, dict):
            yield from OmegaConfigLoader._iter_nested_dicts(
                raw_value, resolved_node.get(key)
            )

@typing.no_type_check
def _merge_and_resolve(
    self, key: str, config_per_file: dict[Path, DictConfig]
) -> dict[str, Any]:
    """Merge pre-loaded per-file configs and resolve interpolations.

    See ``_resolve_from_raw_config`` for the ``runtime_params`` locking note.
    """
    aggregate_config = config_per_file.values()

    if not aggregate_config:
        return {}

    if key == "parameters":
        # Merge with runtime parameters only for "parameters"
        return OmegaConf.to_container(
            OmegaConf.unsafe_merge(*aggregate_config, self.runtime_params),
            resolve=True,
        )

    merged_config_container = (
        self._resolve_catalog_config(aggregate_config)
        if key == "catalog"
        else OmegaConf.to_container(
            OmegaConf.unsafe_merge(*aggregate_config), resolve=True
        )
    )

    return {
        k: v for k, v in merged_config_container.items() if not k.startswith("_")
    }

def _merge_configs(
    self,
    config: dict[str, Any],
    env_config: dict[str, Any],
    key: str,
    env_path: str,
) -> Any:
    merging_strategy = self.merge_strategy.get(key, "destructive")
    try:
        strategy = MergeStrategies[merging_strategy.upper()]

        # Get the corresponding merge function and call it
        merge_function_name = MERGING_IMPLEMENTATIONS[strategy]
        merge_function = getattr(self, merge_function_name)
        return merge_function(config, env_config, env_path)
    except KeyError:
        allowed_strategies = [strategy.name.lower() for strategy in MergeStrategies]
        raise ValueError(
            f"Merging strategy {merging_strategy} not supported. The accepted merging "
            f"strategies are {allowed_strategies}."
        )

@typing.no_type_check
def _read_dir_configs(
    self,
    conf_path: str,
    patterns: Iterable[str],
    key: str,
    processed_files: set,
    read_environment_variables: bool | None = False,
) -> dict[Path, DictConfig]:
    """Discover, load, and dedupe raw config files from ``conf_path``. No merge, no resolve."""
    # Handle directory existence check for remote paths
    if self._protocol in CLOUD_PROTOCOLS or self._protocol in HTTP_PROTOCOLS:
        try:
            # Check directory existence in remote paths
            self._fs.ls(conf_path)
        except Exception as exc:
            raise MissingConfigException(
                f"Given configuration path either does not exist "
                f"or is not a valid directory: {conf_path}. Error: {exc!s}"
            )
    # Original check for local paths
    elif not self._fs.isdir(Path(conf_path).as_posix()):
        raise MissingConfigException(
            f"Given configuration path either does not exist "
            f"or is not a valid directory: {conf_path}"
        )

    paths = []
    for pattern in patterns:
        for each in self._fs.glob(Path(f"{conf_path!s}/{pattern}").as_posix()):
            if not self.ignore_hidden or not self._is_hidden(each):
                paths.append(Path(each))

    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
            cursor = exc.problem_mark.column
            raise ParserError(
                f"Invalid YAML or JSON file {Path(config_filepath).as_posix()},"
                f" unable to read line {line}, position {cursor}."
            ) from exc

    self._check_duplicates(key, config_per_file)
    return config_per_file

def _read_raw_config(
    self, key: str
) -> tuple[dict[Path, DictConfig], dict[Path, DictConfig], str, str, set[Path]]:
    """Read raw per-file configs from base and env dirs for ``key``.

    Split out so callers can cache the parse and only re-pay the
    resolve cost per request.
    """
    patterns = [*self.config_patterns[key]]
    read_environment_variables = key == "credentials"

    processed_files: set[Path] = set()
    base_path = self._get_conf_env_path(self.base_env)
    base_configs = self._read_dir_configs(
        base_path, patterns, key, processed_files, read_environment_variables
    )

    run_env = self.env or self.default_run_env
    env_path = self._get_conf_env_path(run_env)
    if run_env == self.base_env:
        return base_configs, {}, base_path, env_path, processed_files

    env_configs = self._read_dir_configs(
        env_path, patterns, key, processed_files, read_environment_variables
    )
    return base_configs, env_configs, base_path, env_path, processed_files

def _register_globals_resolver(self) -> None:
    """Register the globals resolver"""
    OmegaConf.register_new_resolver(
        "globals",
        self._get_globals_value,
        replace=True,
    )

@staticmethod
def _register_new_resolvers(resolvers: dict[str, Callable]) -> None:
    """Register custom resolvers"""
    for name, resolver in resolvers.items():
        if not OmegaConf.has_resolver(name):
            msg = f"Registering new custom resolver: {name}"
            _config_logger.debug(msg)
            OmegaConf.register_new_resolver(name=name, resolver=resolver)

def _register_runtime_params_resolver(self) -> None:
    OmegaConf.register_new_resolver(
        "runtime_params",
        self._get_runtime_value,
        replace=True,
    )

@typing.no_type_check
def _resolve_catalog_config(
    self, aggregate_config: Iterable[DictConfig]
) -> dict[str, Any]:
    """Resolve the catalog config, then guard against `runtime_params:`
    selecting a dataset's `type` -- see `_guard_runtime_params_in_catalog_type`.
    """
    # Scope hit-tracking to this resolution pass only.
    self._runtime_params_hits = set()

    merged_config_container = OmegaConf.to_container(
        OmegaConf.unsafe_merge(*aggregate_config), resolve=True
    )
    self._guard_runtime_params_in_catalog_type(
        aggregate_config, merged_config_container
    )
    return merged_config_container

@staticmethod
def _resolve_environment_variables(config: DictConfig) -> None:
    """Use the ``oc.env`` resolver to read environment variables and replace
    them in-place, clearing the resolver after the operation is complete if
    it was not registered beforehand.

    Args:
        config (DictConfig): The configuration dictionary to resolve.
    """
    if not OmegaConf.has_resolver("oc.env"):
        OmegaConf.register_new_resolver("oc.env", oc.env)
        OmegaConf.resolve(config)
        OmegaConf.clear_resolver("oc.env")
    else:
        OmegaConf.resolve(config)

def _resolve_from_raw_config(  # noqa: PLR0913
    self,
    key: str,
    base_configs: dict[Path, DictConfig],
    env_configs: dict[Path, DictConfig],
    base_path: str,
    env_path: str,
    processed_files: set[Path],
) -> dict[str, Any]:
    """Merge and resolve pre-parsed per-file configs into the final dict.

    Uses ``self.runtime_params`` as currently set. Resolver registrations
    are process-global, so callers resolving per-request params must set
    them under a lock before calling.
    """
    if key == "globals":
        # "runtime_params" resolver is not allowed in globals.
        OmegaConf.clear_resolver("runtime_params")

    try:
        base_config = self._merge_and_resolve(key, base_configs)
    except UnsupportedInterpolationType as exc:
        if "runtime_params" in str(exc):
            raise UnsupportedInterpolationType(
                "The `runtime_params:` resolver is not supported for globals."
            )
        raise

    run_env = self.env or self.default_run_env
    if run_env == self.base_env:
        return base_config  # type: ignore[no-any-return]

    try:
        env_config = self._merge_and_resolve(key, env_configs)
    except UnsupportedInterpolationType as exc:
        if "runtime_params" in str(exc):
            raise UnsupportedInterpolationType(
                "The `runtime_params:` resolver is not supported for globals."
            )
        raise

    resulting_config = self._merge_configs(base_config, env_config, key, env_path)

    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 resulting_config  # type: ignore[no-any-return]

@staticmethod
def _soft_merge(
    config: dict[str, Any], env_config: dict[str, Any], env_path: str | None = None
) -> Any:
    # Soft merge the two env dirs. The chosen env will override base if keys clash.
    return OmegaConf.to_container(OmegaConf.unsafe_merge(config, env_config))

def keys(self) -> KeysView:
    return KeysView(self.config_patterns)

@typing.no_type_check
def load_and_merge_dir_config(
    self,
    conf_path: str,
    patterns: Iterable[str],
    key: str,
    processed_files: set,
    read_environment_variables: bool | None = False,
) -> dict[str, Any]:
    """Recursively load and merge all configuration files in a directory using OmegaConf,
    which satisfy a given list of glob patterns from a specific path.

    Args:
        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.

    """
    config_per_file = self._read_dir_configs(
        conf_path, patterns, key, processed_files, read_environment_variables
    )
    return self._merge_and_resolve(key, config_per_file)