Catalog to config

RELEASE.md

@@ -1,7 +1,11 @@
 # Upcoming Release
 
 ## Major features and improvements
+* Implemented `KedroDataCatalog.to_config()` method that converts the catalog instance into a configuration format suitable for serialization.
+
 ## Bug fixes and other changes
+* Added validation to ensure dataset versions consistency across catalog.
+
 ## Breaking changes to the API
 ## Documentation changes
 ## Community contributions

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 move it to 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

@@ -30,7 +30,7 @@ def __init__(
         self._dataset_patterns, self._default_pattern = self._extract_patterns(
             config, credentials
         )
-        self._resolved_configs = self._resolve_config_credentials(config, credentials)
+        self._resolved_configs = self.resolve_credentials(config, credentials)
 
     @property
     def config(self) -> dict[str, dict[str, Any]]:
@@ -237,8 +237,9 @@ def _extract_patterns(
 
         return sorted_patterns, user_default
 
-    def _resolve_config_credentials(
-        self,
+    @classmethod
+    def resolve_credentials(
+        cls,
         config: dict[str, dict[str, Any]] | None,
         credentials: dict[str, dict[str, Any]] | None,
     ) -> dict[str, dict[str, Any]]:
@@ -254,13 +255,55 @@ 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_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}"
+
+        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):
@@ -71,16 +73,16 @@ class DatasetError(Exception):
 
 
 class DatasetNotFoundError(DatasetError):
-    """``DatasetNotFoundError`` raised by ``DataCatalog`` class in case of
-    trying to use a non-existing dataset.
+    """``DatasetNotFoundError`` raised by ```DataCatalog`` and ``KedroDataCatalog``
+    classes in case of trying to use a non-existing dataset.
     """
 
     pass
 
 
 class DatasetAlreadyExistsError(DatasetError):
-    """``DatasetAlreadyExistsError`` raised by ``DataCatalog`` class in case
-    of trying to add a dataset which already exists in the ``DataCatalog``.
+    """``DatasetAlreadyExistsError`` raised by ```DataCatalog`` and ``KedroDataCatalog``
+    classes in case of trying to add a dataset which already exists in the ``DataCatalog``.
     """
 
     pass
@@ -94,6 +96,15 @@ class VersionNotFoundError(DatasetError):
     pass
 
 
+class VersionAlreadyExistsError(DatasetError):
+    """``VersionAlreadyExistsError`` raised by ``DataCatalog`` and ``KedroDataCatalog``
+    classes when attempting to add a dataset to a catalog with a save version
+    that conflicts with the save version already set for the catalog.
+    """
+
+    pass
+
+
 _DI = TypeVar("_DI")
 _DO = TypeVar("_DO")
 
@@ -148,6 +159,23 @@ class AbstractDataset(abc.ABC, Generic[_DI, _DO]):
     need to change the `_EPHEMERAL` attribute to 'True'.
     """
     _EPHEMERAL = False
+    # Declares a class-level attribute that will store the initialization
+    # arguments of an instance. Initially, it is set to None, but it will
+    # hold a dictionary of arguments after initialization.
+    _init_args: dict[str, Any] | None = None
+
+    def __post_init__(self, call_args: dict[str, Any]) -> None:
+        """Handles additional setup after the object is initialized.
+
+        Stores the initialization arguments (excluding `self`) in the `_init_args` attribute.
+
+        Args:
+            call_args: A dictionary of arguments passed to the `__init__` method, captured
+                using `inspect.getcallargs`.
+        """
+
+        self._init_args = call_args
+        self._init_args.pop("self", None)
 
     @classmethod
     def from_config(
@@ -201,6 +229,58 @@ def from_config(
             ) from err
         return dataset
 
+    def to_config(self) -> dict[str, Any]:
+        """Converts the dataset instance into a dictionary-based configuration for
+        serialization. Ensures that any subclass-specific details are handled, with
+        additional logic for versioning and caching implemented for `CachedDataset`.
+
+        Adds a key for the dataset's type using its module and class name and
+        includes the initialization arguments.
+
+        For `CachedDataset` it extracts the underlying dataset's configuration,
+        handles the `versioned` flag and removes unnecessary metadata. It also
+        ensures the embedded dataset's configuration is appropriately flattened
+        or transformed.
+
+        If the dataset has a version key, it sets the `versioned` flag in the
+        configuration.
+
+        Removes the `metadata` key from the configuration if present.
+
+        Returns:
+            A dictionary containing the dataset's type and initialization arguments.
+        """
+        return_config: dict[str, Any] = {
+            f"{TYPE_KEY}": f"{type(self).__module__}.{type(self).__name__}"
+        }
+
+        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 = 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
+                )
+            # Pop metadata from configuration
+            cached_ds_return_config.pop("metadata", None)
+            return_config["dataset"] = cached_ds_return_config
+
+        # Set `versioned` key if version present in the dataset
+        if return_config.pop(VERSION_KEY, None):
+            return_config[VERSIONED_FLAG_KEY] = True
+
+        # Pop metadata from configuration
+        return_config.pop("metadata", None)
+
+        return return_config
+
     @property
     def _logger(self) -> logging.Logger:
         return logging.getLogger(__name__)
@@ -281,11 +361,49 @@ def save(self: Self, data: _DI) -> None:
         return save
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
-        """Decorate the `load` and `save` methods provided by the class.
+        """Customizes the behavior of subclasses of AbstractDataset during
+        their creation. This method is automatically invoked when a subclass
+        of AbstractDataset is defined.
+
 
+        Decorates the `load` and `save` methods provided by the class.
         If `_load` or `_save` are defined, alias them as a prerequisite.
 
         """
+
+        # Save the original __init__ method of the subclass
+        init_func: Callable = cls.__init__
+
+        def init_decorator(previous_init: Callable) -> Callable:
+            """A decorator that wraps the original __init__ of the subclass.
+
+            It ensures that after the original __init__ executes, the `__post_init__`
+            method of the instance is called with the arguments used to initialize
+            the object.
+            """
+
+            def new_init(self, *args, **kwargs) -> None:  # type: ignore[no-untyped-def]
+                """The decorated __init__ method.
+
+                Executes the original __init__, then calls __post_init__ with the arguments
+                used to initialize the instance.
+                """
+
+                # Call the original __init__ method
+                previous_init(self, *args, **kwargs)
+                if type(self) is cls:
+                    # Capture and process the arguments passed to the original __init__
+                    call_args = getcallargs(init_func, self, *args, **kwargs)
+                    # Call the custom post-initialization method to save captured arguments
+                    self.__post_init__(call_args)
+
+            return new_init
+
+        # Replace the subclass's __init__ with the decorated version
+        # A hook for subclasses to capture initialization arguments and save them
+        # in the AbstractDataset._init_args field
+        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 +602,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 +1073,67 @@ 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 _validate_versions(
+    datasets: dict[str, AbstractDataset] | None,
+    load_versions: dict[str, str],
+    save_version: str | None,
+) -> tuple[dict[str, str], str | None]:
+    """Validates and synchronises dataset versions for loading and saving.
+
+    Ensures consistency of dataset versions across a catalog, particularly
+    for versioned datasets. It updates load versions and validates that all
+    save versions are consistent.
+
+    Args:
+        datasets: A dictionary mapping dataset names to their instances.
+            if None, no validation occurs.
+        load_versions: A mapping between dataset names and versions
+            to load.
+        save_version: Version string to be used for ``save`` operations
+            by all datasets with versioning enabled.
+
+    Returns:
+        Updated ``load_versions`` with load versions specified in the ``datasets``
+            and resolved ``save_version``.
+
+    Raises:
+        VersionAlreadyExistsError: If a dataset's save version conflicts with
+            the catalog's save version.
+    """
+    if not datasets:
+        return load_versions, save_version
+
+    cur_load_versions = load_versions.copy()
+    cur_save_version = save_version
+
+    for ds_name, ds in datasets.items():
+        # TODO: Move to kedro/io/kedro_data_catalog.py when removing DataCatalog
+        # TODO: Make it a protected static method for KedroDataCatalog
+        # TODO: Replace with isinstance(ds, CachedDataset) - current implementation avoids circular import
+        cur_ds = ds._dataset if ds.__class__.__name__ == "CachedDataset" else ds  # type: ignore[attr-defined]
+
+        if isinstance(cur_ds, AbstractVersionedDataset) and cur_ds._version:
+            if cur_ds._version.load:
+                cur_load_versions[ds_name] = cur_ds._version.load
+            if cur_ds._version.save:
+                cur_save_version = cur_save_version or cur_ds._version.save
+                if cur_save_version != cur_ds._version.save:
+                    raise VersionAlreadyExistsError(
+                        f"Cannot add a dataset `{ds_name}` with `{cur_ds._version.save}` save version. "
+                        f"Save version set for the catalog is `{cur_save_version}`"
+                        f"All datasets in the catalog must have the same save version."
+                    )
+
+    return cur_load_versions, cur_save_version
+
+
+def _is_memory_dataset(ds_or_type: AbstractDataset | str) -> bool:
+    """Check if dataset or str type provided is a MemoryDataset."""
+    if isinstance(ds_or_type, AbstractDataset):
+        return ds_or_type.__class__.__name__ == "MemoryDataset"
+    if isinstance(ds_or_type, str):
+        return ds_or_type in {"MemoryDataset", "kedro.io.memory_dataset.MemoryDataset"}
+
+    return False