Repository: kedro-org/kedro
File: kedro/config/omegaconf_config.py (Python)

Code before:
                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

        if key == "parameters":
            seen_file_to_keys = {
                file: self._get_all_keys(OmegaConf.to_container(config, resolve=False))
                for file, config in config_per_file.items()
            }
        else:
            seen_file_to_keys = {
                file: set(config.keys()) for file, config in config_per_file.items()
            }
        aggregate_config = config_per_file.values()
        self._check_duplicates(seen_file_to_keys)

        if not aggregate_config:
            return {}

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

        return {
            k: v
            for k, v in OmegaConf.to_container(
                OmegaConf.merge(*aggregate_config), resolve=True
            ).items()
            if not k.startswith("_")
        }

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

        for key, value in cfg.items():
            key_str = str(key)

Review comment (refactor):
The logic is all fine, but this is making `load_and_merge_dir_config` even longer.. I'd suggest refactoring this by just be adding this logic to `_check_duplicates`. 

Code after:
            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

        aggregate_config = config_per_file.values()
        self._check_duplicates(key, config_per_file)

        if not aggregate_config:
            return {}

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

        return {
            k: v
            for k, v in OmegaConf.to_container(
                OmegaConf.merge(*aggregate_config), resolve=True
            ).items()
            if not k.startswith("_")
        }

    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 _is_valid_config_path(self, path: Path) -> bool:
        """Check if given path is a file path and file type is yaml or json."""