Catalog to config

kedro/framework/cli/catalog.py

@@ -28,6 +28,7 @@ def _create_session(package_name: str, **kwargs: Any) -> KedroSession:
 
 
 def is_parameter(dataset_name: str) -> bool:
+    # TODO: when breaking change replace with is_parameter from kedro/io/core.py
     """Check if dataset is a parameter."""
     return dataset_name.startswith("params:") or dataset_name == "parameters"
 

kedro/io/catalog_config_resolver.py

@@ -237,8 +237,9 @@ def _extract_patterns(
 
         return sorted_patterns, user_default
 
+    @classmethod
     def _resolve_config_credentials(
-        self,
+        cls,
         config: dict[str, dict[str, Any]] | None,
         credentials: dict[str, dict[str, Any]] | None,
     ) -> dict[str, dict[str, Any]]:
@@ -254,13 +255,36 @@ def _resolve_config_credentials(
                     "\nHint: If this catalog entry is intended for variable interpolation, "
                     "make sure that the key is preceded by an underscore."
                 )
-            if not self.is_pattern(ds_name):
-                resolved_configs[ds_name] = self._resolve_credentials(
+            if not cls.is_pattern(ds_name):
+                resolved_configs[ds_name] = cls._resolve_credentials(
                     ds_config, credentials
                 )
 
         return resolved_configs
 
+    @staticmethod
+    def unresolve_config_credentials(
+        cred_name: str, ds_config: dict[str, dict[str, Any]] | None
+    ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
+        ds_config_copy = copy.deepcopy(ds_config) or {}
+        credentials: dict[str, Any] = {}
+        credentials_ref = f"{cred_name}_{CREDENTIALS_KEY}"
+
+        def unresolve(config: Any) -> None:
+            # We don't expect credentials key appears more than once within the same dataset config,
+            # So once we found the key first time we unresolve it and stop iterating after
+            for key, val in config.items():
+                if key == CREDENTIALS_KEY and config[key]:
+                    credentials[credentials_ref] = config[key]
+                    config[key] = credentials_ref
+                    return
+                if isinstance(val, dict):
+                    unresolve(val)
+
+        unresolve(ds_config_copy)
+
+        return ds_config_copy, credentials
+
     def resolve_pattern(self, ds_name: str) -> dict[str, Any]:
         """Resolve dataset patterns and return resolved configurations based on the existing patterns."""
         matched_pattern = self.match_pattern(ds_name)

kedro/io/core.py

@@ -15,6 +15,7 @@
 from datetime import datetime, timezone
 from functools import partial, wraps
 from glob import iglob
+from inspect import getcallargs
 from operator import attrgetter
 from pathlib import Path, PurePath, PurePosixPath
 from typing import (
@@ -57,6 +58,7 @@
     "s3a",
     "s3n",
 )
+TYPE_KEY = "type"
 
 
 class DatasetError(Exception):
@@ -148,6 +150,11 @@ class AbstractDataset(abc.ABC, Generic[_DI, _DO]):
     need to change the `_EPHEMERAL` attribute to 'True'.
     """
     _EPHEMERAL = False
+    _init_args: dict[str, Any] | None = None
+
+    def __post_init__(self, call_args: dict[str, Any]) -> None:
+        self._init_args = call_args
+        self._init_args.pop("self", None)
 
     @classmethod
     def from_config(
@@ -201,6 +208,49 @@ def from_config(
             ) from err
         return dataset
 
+    def to_config(self) -> tuple[dict[str, Any], dict[str, str] | None, str | None]:
+        return_config: dict[str, Any] = {
+            f"{TYPE_KEY}": f"{type(self).__module__}.{type(self).__name__}"
+        }
+        load_versions: dict[str, str] | None = None
+        save_version: str | None = None
+
+        if self._init_args:
+            return_config.update(self._init_args)
+
+        if type(self).__name__ == "CachedDataset":
+            cached_ds = return_config.pop("dataset")
+            cached_ds_return_config: dict[str, Any] = {}
+            if isinstance(cached_ds, dict):
+                cached_ds_return_config = cached_ds
+            elif isinstance(cached_ds, AbstractDataset):
+                cached_ds_return_config, load_versions, save_version = (
+                    cached_ds.to_config()
+                )
+            if VERSIONED_FLAG_KEY in cached_ds_return_config:
+                return_config[VERSIONED_FLAG_KEY] = cached_ds_return_config.pop(
+                    VERSIONED_FLAG_KEY
+                )
+            return_config["dataset"] = cached_ds_return_config
+
+        version = return_config.pop(VERSION_KEY, None)
+
+        if version:
+            return_config[VERSIONED_FLAG_KEY] = True
+            load_versions, save_version = (
+                load_versions or version.load,
+                save_version or version.save,
+            )
+
+        # Pop data from configuration
+        if type(self).__name__ == "MemoryDataset":
+            return_config.pop("data", None)
+
+        # Pop metadata from configuration
+        return_config.pop("metadata", None)
+
+        return return_config, load_versions, save_version
+
     @property
     def _logger(self) -> logging.Logger:
         return logging.getLogger(__name__)
@@ -286,6 +336,20 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
         If `_load` or `_save` are defined, alias them as a prerequisite.
 
         """
+
+        init_func: Callable = cls.__init__
+
+        def init_decorator(previous_init: Callable) -> Callable:
+            def new_init(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
+                previous_init(self, *args, **kwargs)
+                if type(self) is cls:
+                    call_args = getcallargs(init_func, self, *args, **kwargs)
+                    self.__post_init__(call_args)
+
+            return new_init
+
+        cls.__init__ = init_decorator(cls.__init__)  # type: ignore[method-assign]
+
         super().__init_subclass__(**kwargs)
 
         if hasattr(cls, "_load") and not cls._load.__qualname__.startswith("Abstract"):
@@ -484,14 +548,14 @@ def parse_dataset_definition(
     config = copy.deepcopy(config)
 
     # TODO: remove when removing old catalog as moved to KedroDataCatalog
-    if "type" not in config:
+    if TYPE_KEY not in config:
         raise DatasetError(
             "'type' is missing from dataset catalog configuration."
             "\nHint: If this catalog entry is intended for variable interpolation, "
             "make sure that the top level key is preceded by an underscore."
         )
 
-    dataset_type = config.pop("type")
+    dataset_type = config.pop(TYPE_KEY)
     class_obj = None
     if isinstance(dataset_type, str):
         if len(dataset_type.strip(".")) != len(dataset_type):
@@ -955,3 +1019,9 @@ def confirm(self, name: str) -> None:
     def shallow_copy(self, extra_dataset_patterns: Patterns | None = None) -> _C:
         """Returns a shallow copy of the current object."""
         ...
+
+
+def _is_parameter(dataset_name: str) -> bool:
+    # TODO: when breaking change replace with is_parameter and remove is_parameter from kedro/framework/cli/catalog.py
+    """Check if dataset is a parameter."""
+    return dataset_name.startswith("params:") or dataset_name == "parameters"

kedro/io/kedro_data_catalog.py

@@ -18,13 +18,15 @@
 
 from kedro.io.catalog_config_resolver import CatalogConfigResolver, Patterns
 from kedro.io.core import (
+    VERSIONED_FLAG_KEY,
     AbstractDataset,
     AbstractVersionedDataset,
     CatalogProtocol,
     DatasetAlreadyExistsError,
     DatasetError,
     DatasetNotFoundError,
     Version,
+    _is_parameter,
     generate_timestamp,
 )
 from kedro.io.memory_dataset import MemoryDataset
@@ -96,7 +98,7 @@ def __init__(
             >>> catalog = KedroDataCatalog(datasets={"cars": cars})
         """
         self._config_resolver = config_resolver or CatalogConfigResolver()
-        self._datasets = datasets or {}
+        self._datasets: dict[str, AbstractDataset] = datasets or {}
         self._lazy_datasets: dict[str, _LazyDataset] = {}
         self._load_versions = load_versions or {}
         self._save_version = save_version
@@ -364,6 +366,51 @@ class to be loaded is specified with the key ``type`` and their
             config_resolver=config_resolver,
         )
 
+    def to_config(
+        self,
+    ) -> tuple[
+        dict[str, dict[str, Any]],
+        dict[str, dict[str, Any]],
+        dict[str, str | None],
+        dict[str, str | None],
+    ]:
+        catalog: dict[str, dict[str, Any]] = {}
+        credentials: dict[str, dict[str, Any]] = {}
+        load_version: dict[str, str | None] = {}
+        save_version: dict[str, str | None] = {}
+
+        for ds_name, ds in self._lazy_datasets.items():
+            if _is_parameter(ds_name):
+                continue
+            unresolved_config, unresolved_credentials = (
+                self._config_resolver.unresolve_config_credentials(ds_name, ds.config)
+            )
+            catalog[ds_name] = unresolved_config
+            credentials.update(unresolved_credentials)
+            # TODO: Update when #4327 resolved
+            if catalog[ds_name].get(VERSIONED_FLAG_KEY, None):
+                load_version[ds_name] = ds.load_version
+                save_version[ds_name] = ds.save_version
+            else:
+                load_version[ds_name] = None
+                save_version[ds_name] = None
+
+        for ds_name, ds in self._datasets.items():  # type: ignore[assignment]
+            if _is_parameter(ds_name):
+                continue
+            resolved_config, cur_load_versions, cur_save_version = ds.to_config()  # type: ignore[attr-defined]
+            unresolved_config, unresolved_credentials = (
+                self._config_resolver.unresolve_config_credentials(
+                    ds_name, resolved_config
+                )
+            )
+            catalog[ds_name] = unresolved_config
+            credentials.update(unresolved_credentials)
+            load_version[ds_name] = cur_load_versions
+            save_version[ds_name] = cur_save_version
+
+        return catalog, credentials, load_version, save_version
+
     @staticmethod
     def _validate_dataset_config(ds_name: str, ds_config: Any) -> None:
         if not isinstance(ds_config, dict):

tests/io/conftest.py

@@ -51,6 +51,52 @@ def correct_config(filepath):
     }
 
 
+@pytest.fixture
+def correct_config_versioned(filepath):
+    return {
+        "catalog": {
+            "boats": {
+                "type": "pandas.CSVDataset",
+                "filepath": filepath,
+                "versioned": True,
+            },
+            "cars": {
+                "type": "pandas.CSVDataset",
+                "filepath": "s3://test_bucket/test_file.csv",
+                "credentials": "cars_credentials",
+            },
+            "cars_ibis": {
+                "type": "ibis.FileDataset",
+                "filepath": "cars_ibis.csv",
+                "file_format": "csv",
+                "table_name": "cars",
+                "connection": {"backend": "duckdb", "database": "company.db"},
+                "load_args": {"sep": ",", "nullstr": "#NA"},
+                "save_args": {"sep": ",", "nullstr": "#NA"},
+            },
+            "cached_ds": {
+                "type": "kedro.io.cached_dataset.CachedDataset",
+                "versioned": True,
+                "dataset": {
+                    "type": "pandas.CSVDataset",
+                    "filepath": "cached_ds.csv",
+                    "credentials": "cached_ds_credentials",
+                },
+                "copy_mode": None,
+            },
+            "parameters": {
+                "type": "kedro.io.memory_dataset.MemoryDataset",
+                "data": [4, 5, 6],
+                "copy_mode": None,
+            },
+        },
+        "credentials": {
+            "cars_credentials": {"key": "FAKE_ACCESS_KEY", "secret": "FAKE_SECRET_KEY"},
+            "cached_ds_credentials": {"key": "KEY", "secret": "SECRET"},
+        },
+    }
+
+
 @pytest.fixture
 def correct_config_with_nested_creds(correct_config):
     correct_config["catalog"]["cars"]["credentials"] = {

tests/io/test_catalog_config_resolver.py

@@ -0,0 +1,38 @@
+from kedro.io import CatalogConfigResolver
+
+
+class TestCatalogConfigResolver:
+    def test_unresolve_config_credentials(self, correct_config):
+        """Test unresolve dataset credentials to original format."""
+        config = correct_config["catalog"]
+        credentials = correct_config["credentials"]
+        resolved_configs = CatalogConfigResolver._resolve_config_credentials(
+            config, credentials
+        )
+
+        unresolved_config, unresolved_credentials = (
+            CatalogConfigResolver.unresolve_config_credentials(
+                cred_name="s3", ds_config=resolved_configs
+            )
+        )
+        assert config == unresolved_config
+        assert credentials == unresolved_credentials
+
+    def test_unresolve_config_credentials_two_keys(self, correct_config):
+        """Test unresolve dataset credentials to original format when two credentials keys provided."""
+        config = correct_config["catalog"]
+        credentials = correct_config["credentials"]
+
+        resolved_configs = CatalogConfigResolver._resolve_config_credentials(
+            config, credentials
+        )
+        resolved_configs["cars"]["metadata"] = {"credentials": {}}
+
+        unresolved_config, unresolved_credentials = (
+            CatalogConfigResolver.unresolve_config_credentials(
+                cred_name="s3", ds_config=resolved_configs
+            )
+        )
+        unresolved_config["cars"].pop("metadata")
+        assert config == unresolved_config
+        assert credentials == unresolved_credentials