Repository: kedro-org/kedro
File: kedro/io/catalog_config_resolver.py (Python)

Code before:

        return sorted_patterns, user_default

    @classmethod
    def resolve_credentials(
        cls,
        config: dict[str, dict[str, Any]] | None,
        credentials: dict[str, dict[str, Any]] | None,
    ) -> dict[str, dict[str, Any]]:
        """Initialize the dataset configuration with resolved credentials."""
        config = config or {}
        credentials = credentials or {}
        resolved_configs = {}

        for ds_name, ds_config in config.items():
            if not isinstance(ds_config, dict):
                raise DatasetError(
                    f"Catalog entry '{ds_name}' is not a valid dataset configuration. "
                    "\nHint: If this catalog entry is intended for variable interpolation, "
                    "make sure that the key is preceded by an underscore."
                )
            if not cls.is_pattern(ds_name):
                resolved_configs[ds_name] = cls._resolve_credentials(
                    ds_config, credentials
                )

        return resolved_configs

    @staticmethod
    def unresolve_credentials(
        cred_name: str, ds_config: dict[str, dict[str, Any]] | None
    ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
        """Extracts and replaces credentials in a dataset configuration with
        references, ensuring separation of credentials from the dataset configuration.

        Credentials are searched for recursively in the dataset configuration.
        The first occurrence of the `CREDENTIALS_KEY` is replaced with a generated
        reference key.

        Args:
            cred_name: A unique identifier for the credentials being unresolved.
                This is used to generate a reference key for the credentials.
            ds_config: The dataset configuration containing potential credentials
                under the key `CREDENTIALS_KEY`.

        Returns:
            A tuple containing:
                ds_config_copy : A deep copy of the original dataset
                    configuration with credentials replaced by reference keys.
                credentials: A dictionary mapping generated reference keys to the original credentials.
        """

Review comment (security):
[Nit] For me the name is a bit hard to understand what the method is doing. We are replacing `CREDENTIALS_KEY` with a generated ref key to differentiate credentials from ds config. Would `_replace_credentials_key_with_ref` or `_convert_credentials_to_refs` be more clear ?

Code after:

    @classmethod
    def _resolve_credentials(
        cls,
        config: dict[str, dict[str, Any]] | None,
        credentials: dict[str, dict[str, Any]] | None,
    ) -> dict[str, dict[str, Any]]:
        """Initialize the dataset configuration with resolved credentials."""
        config = config or {}
        credentials = credentials or {}
        resolved_configs = {}

        for ds_name, ds_config in config.items():
            if not isinstance(ds_config, dict):
                raise DatasetError(
                    f"Catalog entry '{ds_name}' is not a valid dataset configuration. "
                    "\nHint: If this catalog entry is intended for variable interpolation, "
                    "make sure that the key is preceded by an underscore."
                )
            if not cls.is_pattern(ds_name):
                resolved_configs[ds_name] = _resolve_credentials(ds_config, credentials)

        return resolved_configs

    @staticmethod
    def _unresolve_credentials(
        cred_name: str, ds_config: dict[str, dict[str, Any]] | None
    ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
        """Extracts and replaces credentials in a dataset configuration with
        references, ensuring separation of credentials from the dataset configuration.

        Credentials are searched for recursively in the dataset configuration.
        The first occurrence of the `CREDENTIALS_KEY` is replaced with a generated
        reference key.

        Args:
            cred_name: A unique identifier for the credentials being unresolved.
                This is used to generate a reference key for the credentials.
            ds_config: The dataset configuration containing potential credentials
                under the key `CREDENTIALS_KEY`.

        Returns:
            A tuple containing:
                ds_config_copy : A deep copy of the original dataset
                    configuration with credentials replaced by reference keys.
                credentials: A dictionary mapping generated reference keys to the original credentials.
        """
        ds_config_copy = copy.deepcopy(ds_config) or {}
        credentials: dict[str, Any] = {}
        credentials_ref = f"{cred_name}_{CREDENTIALS_KEY}"