Repository: kedro-org/kedro
Pull request: Make dataset factory resolve nested dict properly
## Description
<!-- Why was this PR created? -->
Fix #2992 
- resolve config properly for nested dict
- resolve config properly for lists

## Development notes
<!-- What have you changed, and how has this been tested? -->
* I have updated the `DataCatalog._resolve_config()` function -
    * To recursively call itself if the config is a nested dict.
    * Recursively call itself for each item if the config is a list
    * if it is a string, try to replace the placeholder
    * Into a class method since it doesn't need access to `DataCatalog._dataset_patterns` but takes the config dict as an argument
* Added a test for nested dict config, list, list + dict cases
* Updated dataset factories CLI commands to use the updated `_resolve_config()`
* Updated release notes
## Checklist

- [x] Read the [contributing](https://github.com/kedro-org/kedro/blob/main/CONTRIBUTING.md) guidelines
- [x] Opened this PR as a 'Draft Pull Request' if it is work-in-progress
- [ ] Updated the documentation to reflect the code changes
- [x] Added a description of this change in the [`RELEASE.md`](https://github.com/kedro-org/kedro/blob/main/RELEASE.md) file
- [x] Added tests to cover my changes
- [ ] Checked if this change will affect Kedro-Viz, and if so, communicated that with the Viz team


Problem statement:
Dataset factories doesn't resolve nested config properly
## Description
Reported by @m-gris
Dataset factories does not work properly for config that is nested.

## Context
The problem is in `DataCatalog._resolve_config()`. It fills out the placeholder on top level keys properly but doesn't go into nested dicts.

## Steps to Reproduce
Message quoted from slack - 

> Hi everyone,
> I’ve started to use kedro-mlflow and am under the impression that there is some conflict / incompatibility with the dataset factory.
> Having moved from
> ```yaml
> "models.{model}.embeddings":
>     type: pickle.PickleDataSet
>     filepath: "data/07_model_output/{model}_embeddings.pkl"
>     backend: joblib
>     versioned: True
>     layer: model_output
> ```
> to
> ```yaml
> "models.{model}.embeddings":
>   type: kedro_mlflow.io.artifacts.MlflowArtifactDataSet
>   data_set:
>     type: pickle.PickleDataSet
>     filepath: "data/07_model_output/{model}_embeddings.pkl"
>     backend: joblib
>     versioned: True
>   layer: model_output
> ```
> results in artifacts being saved without the model named properly interpolated, i.e “litteraly” saved as {model}_embeddings.pkl
> Any comment ? Anything I’m missing ?
>

Hints from issue discussion:
Hi @ankatiyar, 
I encountered the same problem and was about to post an issue when I found yours. 
I think the config resolver also fails if an argument contains a list (not only a nested dict).
I was about to suggest something like this:

```python 
    def _resolve_config(
        self,
        data_set_name: str,
        matched_pattern: str,
    ) -> dict[str, Any]:
        """Get resolved AbstractDataset from a factory config"""
        result = parse(matched_pattern, data_set_name)
        config_copy = copy.deepcopy(self._dataset_patterns[matched_pattern])
        config_copy = self._recursively_resolve_nested_fields(
            config_copy,
            data_set_name,
            matched_pattern, 
            result)
        return config_copy

    def _recursively_resolve_nested_fields(
            self,
            config_field,
            data_set_name,
            matched_pattern, 
            result)-> dict[str, Any]:
        if isinstance(config_field, dict):
            for key, value in config_field.items():
                config_field[key] = self._recursively_resolve_nested_fields(
                    value,
                    data_set_name,
                    matched_pattern,
                    result
                    )
        elif isinstance(config_field, list):
            for i, value in enumerate(config_field):
                config_field[i] = self._recursively_resolve_nested_fields(
                    value,
                    data_set_name,
                    matched_pattern,
                    result
                    )
        elif isinstance(config_field, Iterable) and "}" in config_field:
            # result.named: gives access to all dict items in the match result.
            # format_map fills in dict values into a string with {...} placeholders
            # of the same key name.
            try:
                config_field = str(config_field).format_map(result.named)
            except KeyError as exc:
                raise DatasetError(
                    f"Unable to resolve '{config_field}' for the pattern '{matched_pattern}'"
                ) from exc
        return config_field
```

Do you think the list case should be added to the PR? 



Hi @Gabriel2409,
Thanks for flagging this and suggesting a solution for this too! 😄 
I'll try fixing this in the same open PR. 
Would you mind providing an example of what a catalog entry with a list would look like? 
@ankatiyar 
For example, in pandas.CSVDataSet, load_args.na_values is a list

So we could have something like this, which gives you different ways to load the same file by specifying a different column as NA (a bit artificial I admit but it is for the sake of the example) 
```yml
"project_ignore_{param}":
    type: pandas.CSVDataSet
    filepath: data/01_raw/project.csv
    load_args:
      sep: ","
      na_values: ["#NA", {param}]
    save_args:
      index: False
      date_format: "%Y-%m-%d %H:%M"
      decimal: .
```


A more straightforward example would be a custom dataset containing a list of filepath.
```yml
myname:
    type: MyCustomDataSet
    filepaths: [path1, path2]
```


Note that in my suggested implementation, if a list contains a dict, only the value is parsed, not the key, which is consistent with how it is currently done

```yml
"my{param}"
    type: MyCustomDataSet
    mylist: [path/to/my{param}, myawesome{param}] # correctly replaced
    mylistofdict: [{"key{param}": "value{param}"}] # {param} only replaced in the value, not the key
```

Commit message: Remove iterable

Signed-off-by: Ankita Katiyar <ankitakatiyar2401@gmail.com>

Patch to review:
diff --git a/RELEASE.md b/RELEASE.md
index 8d5cb62415..6f18db1fe1 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -11,6 +11,8 @@
 
 ## Major features and improvements
 ## Bug fixes and other changes
+* Updated dataset factories to resolve nested catalog config properly.
+
 ## Documentation changes
 ## Breaking changes to the API
 ## Upcoming deprecations for Kedro 0.19.0
diff --git a/kedro/framework/cli/catalog.py b/kedro/framework/cli/catalog.py
index b8849b5843..24816a9492 100644
--- a/kedro/framework/cli/catalog.py
+++ b/kedro/framework/cli/catalog.py
@@ -1,4 +1,5 @@
 """A collection of CLI commands for working with Kedro catalog."""
+import copy
 from collections import defaultdict
 from itertools import chain
 
@@ -84,7 +85,13 @@ def list_datasets(metadata: ProjectMetadata, pipeline, env):
                 data_catalog._dataset_patterns, ds_name
             )
             if matched_pattern:
-                ds_config = data_catalog._resolve_config(ds_name, matched_pattern)
+                ds_config_copy = copy.deepcopy(
+                    data_catalog._dataset_patterns[matched_pattern]
+                )
+
+                ds_config = data_catalog._resolve_config(
+                    ds_name, matched_pattern, ds_config_copy
+                )
                 factory_ds_by_type[ds_config["type"]].append(ds_name)
 
         default_ds = default_ds - set(chain.from_iterable(factory_ds_by_type.values()))
@@ -244,7 +251,14 @@ def resolve_patterns(metadata: ProjectMetadata, env):
             data_catalog._dataset_patterns, ds_name
         )
         if matched_pattern:
-            ds_config = data_catalog._resolve_config(ds_name, matched_pattern)
+            ds_config_copy = copy.deepcopy(
+                data_catalog._dataset_patterns[matched_pattern]
+            )
+
+            ds_config = data_catalog._resolve_config(
+                ds_name, matched_pattern, ds_config_copy
+            )
+
             ds_config["filepath"] = _trim_filepath(
                 str(context.project_path) + "/", ds_config["filepath"]
             )
diff --git a/kedro/io/data_catalog.py b/kedro/io/data_catalog.py
index 4cbe6c0142..815cad8f19 100644
--- a/kedro/io/data_catalog.py
+++ b/kedro/io/data_catalog.py
@@ -11,7 +11,7 @@
 import logging
 import re
 from collections import defaultdict
-from typing import Any, Dict, Iterable
+from typing import Any, Dict
 
 from parse import parse
 
@@ -388,7 +388,10 @@ def _get_dataset(
         if data_set_name not in self._data_sets and matched_pattern:
             # If the dataset is a patterned dataset, materialise it and add it to
             # the catalog
-            data_set_config = self._resolve_config(data_set_name, matched_pattern)
+            config_copy = copy.deepcopy(self._dataset_patterns[matched_pattern])
+            data_set_config = self._resolve_config(
+                data_set_name, matched_pattern, config_copy
+            )
             ds_layer = data_set_config.pop("layer", None)
             if ds_layer:
                 self.layers = self.layers or {}
@@ -436,27 +439,33 @@ def __contains__(self, data_set_name):
             return True
         return False
 
+    @classmethod
     def _resolve_config(
-        self,
+        cls,
         data_set_name: str,
         matched_pattern: str,
+        config: dict,
     ) -> dict[str, Any]:
         """Get resolved AbstractDataset from a factory config"""
         result = parse(matched_pattern, data_set_name)
-        config_copy = copy.deepcopy(self._dataset_patterns[matched_pattern])
         # Resolve the factory config for the dataset
-        for key, value in config_copy.items():
-            if isinstance(value, Iterable) and "}" in value:
-                # result.named: gives access to all dict items in the match result.
-                # format_map fills in dict values into a string with {...} placeholders
-                # of the same key name.
-                try:
-                    config_copy[key] = str(value).format_map(result.named)
-                except KeyError as exc:
-                    raise DatasetError(
-                        f"Unable to resolve '{key}' for the pattern '{matched_pattern}'"
-                    ) from exc
-        return config_copy
+        if isinstance(config, dict):
+            for key, value in config.items():
+                config[key] = cls._resolve_config(data_set_name, matched_pattern, value)
+        elif isinstance(config, (list, tuple)):
+            config = [
+                cls._resolve_config(data_set_name, matched_pattern, value)
+                for value in config
+            ]
+        elif isinstance(config, str) and "}" in config:
+            try:
+                config = str(config).format_map(result.named)
+            except KeyError as exc:
+                raise DatasetError(
+                    f"Unable to resolve '{config}' from the pattern '{matched_pattern}'. Keys used in the config "
+                    f"should be present in the dataset factory pattern."
+                ) from exc
+        return config
 
     def load(self, name: str, version: str = None) -> Any:
         """Loads a registered data set.
diff --git a/tests/io/test_data_catalog.py b/tests/io/test_data_catalog.py
index 9273fa5200..c20abfc6d7 100644
--- a/tests/io/test_data_catalog.py
+++ b/tests/io/test_data_catalog.py
@@ -107,6 +107,29 @@ def config_with_dataset_factories():
     }
 
 
+@pytest.fixture
+def config_with_dataset_factories_nested():
+    return {
+        "catalog": {
+            "{brand}_cars": {
+                "type": "PartitionedDataset",
+                "path": "data/01_raw",
+                "dataset": "pandas.CSVDataSet",
+                "metadata": {
+                    "my-plugin": {
+                        "brand": "{brand}",
+                        "list_config": [
+                            "NA",
+                            "{brand}",
+                        ],
+                        "nested_list_dict": [{}, {"brand": "{brand}"}],
+                    }
+                },
+            },
+        },
+    }
+
+
 @pytest.fixture
 def config_with_dataset_factories_with_default(config_with_dataset_factories):
     config_with_dataset_factories["catalog"]["{default_dataset}"] = {
@@ -840,7 +863,10 @@ def test_unmatched_key_error_when_parsing_config(
     ):
         """Check error raised when key mentioned in the config is not in pattern name"""
         catalog = DataCatalog.from_config(**config_with_dataset_factories_bad_pattern)
-        pattern = "Unable to resolve 'filepath' for the pattern '{type}@planes'"
+        pattern = (
+            "Unable to resolve 'data/01_raw/{brand}_plane.pq' from the pattern '{type}@planes'. "
+            "Keys used in the config should be present in the dataset factory pattern."
+        )
         with pytest.raises(DatasetError, match=re.escape(pattern)):
             catalog._get_dataset("jet@planes")
 
@@ -896,3 +922,13 @@ def test_factory_config_versioned(
             microsecond=current_ts.microsecond // 1000 * 1000, tzinfo=None
         )
         assert actual_timestamp == expected_timestamp
+
+    def test_factory_nested_config(self, config_with_dataset_factories_nested):
+        catalog = DataCatalog.from_config(**config_with_dataset_factories_nested)
+        dataset = catalog._get_dataset("tesla_cars")
+        assert dataset.metadata["my-plugin"]["brand"] == "tesla"
+        assert dataset.metadata["my-plugin"]["list_config"] == ["NA", "tesla"]
+        assert dataset.metadata["my-plugin"]["nested_list_dict"] == [
+            {},
+            {"brand": "tesla"},
+        ]


Code review comments:
diff_hunk: @@ -840,7 +863,10 @@ def test_unmatched_key_error_when_parsing_config(
     ):
         """Check error raised when key mentioned in the config is not in pattern name"""
         catalog = DataCatalog.from_config(**config_with_dataset_factories_bad_pattern)
-        pattern = "Unable to resolve 'filepath' for the pattern '{type}@planes'"
+        pattern = (
+            "Unable to resolve 'data/01_raw/{brand}_plane.pq' from the pattern '{type}@planes'. "
+            "Keys used in the config should be present in the dataset factory pattern."
path: tests/io/test_data_catalog.py
text: @user1:
```suggestion
            "Keys used in the configuration should be present in the dataset factory pattern."
```