Implement catalog filter for KedroDataCatalog

.secrets.baseline

@@ -211,9 +211,9 @@
         "filename": "tests/io/test_kedro_data_catalog.py",
         "hashed_secret": "15dd2c9ccec914f1470b4dccb45789844e49cf70",
         "is_verified": false,
-        "line_number": 499
+        "line_number": 558
       }
     ]
   },
-  "generated_at": "2025-01-27T18:47:13Z"
+  "generated_at": "2025-01-28T15:23:17Z"
 }

RELEASE.md

@@ -5,6 +5,7 @@
 * Improve OmegaConfigLoader performance.
 * Replaced `trufflehog` with `detect-secrets` for detecting secrets within a code base.
 * Added support for `%load_ext kedro`.
+* Implemented `KedroDataCatalog.filter()` to filter datasets by name and type.
 
 ## Bug fixes and other changes
 * Added validation to ensure dataset versions consistency across catalog.
@@ -18,6 +19,7 @@
 * Updated `Partitioned dataset lazy saving` docs page.
 * Fixed `KedroDataCatalog` mutation after pipeline run.
 * Made `KedroDataCatalog._datasets` compatible with `DataCatalog._datasets`.
+* Updated `_LazyDataset` representation when printing `KedroDataCatalog`.
 
 ## Breaking changes to the API
 ## Documentation changes

kedro/io/kedro_data_catalog.py

@@ -27,6 +27,7 @@
     Version,
     _validate_versions,
     generate_timestamp,
+    parse_dataset_definition,
 )
 from kedro.io.memory_dataset import MemoryDataset, _is_memory_dataset
 from kedro.utils import _format_rich, _has_rich_handler
@@ -48,7 +49,8 @@ def __init__(
         self.save_version = save_version
 
     def __repr__(self) -> str:
-        return f"{self.config.get('type', 'UnknownType')}"
+        class_type, _ = parse_dataset_definition(self.config)
+        return f"{class_type.__module__}.{class_type.__qualname__}"
 
     def materialize(self) -> AbstractDataset:
         return AbstractDataset.from_config(
@@ -549,11 +551,82 @@ def add(
             )
         self.__setitem__(ds_name, dataset)
 
+    def filter(
+        self,
+        name_regex: str | None = None,
+        name_regex_flags: int | re.RegexFlag = re.IGNORECASE,
+        type_regex: str | None = None,
+        type_regex_flags: int | re.RegexFlag = 0,
+    ) -> List[str]:  # noqa: UP006
+        """Filter dataset names registered in the catalog based on name and/or type.
+
+        This method allows filtering datasets by their names and/or types using optional
+        regular expression patterns. Each pattern can also support optional regex flags
+        for customization. If no filters are provided, all dataset names are returned.
+
+        Args:
+            name_regex: Optional regular expression to filter dataset names by name.
+            name_regex_flags: Optional regex flags for the name filter.
+                By default, IGNORECASE key is set.
+            type_regex: Optional regular expression to filter dataset names by their type.
+                The provided regex is matched against the full dataset type path, for example:
+                `kedro_datasets.pandas.parquet_dataset.ParquetDataset`.
+            type_regex_flags: Optional regex flags for the type filter.
+
+        Returns:
+            A list of dataset names that match the filtering criteria based on `name_regex`
+            and/or `type_regex`. If no filters are provided, all dataset names are returned.
+
+        Raises:
+            SyntaxError: If the provided regex patterns are invalid.
+
+        Example:
+        ::
+
+            >>> catalog = KedroDataCatalog()
+            >>> # get datasets where the substring 'raw' is present
+            >>> raw_data = catalog.filter(name_regex='raw')
+            >>> # get datasets of a specific type
+            >>> csv_datasets = catalog.filter(type_regex='pandas.excel_dataset.ExcelDataset')
+            >>> # get datasets where names start with 'model_' and are of a specific type
+            >>> model_datasets = catalog.filter(
+            ...     name_regex='^model_',
+            ...     type_regex='ModelDataset',
+            ... )
+        """
+
+        # Apply name filter if specified
+        if name_regex:
+            pattern = _compile_pattern(name_regex, name_regex_flags)
+            filtered_names = [
+                ds_name for ds_name in self.__iter__() if pattern.search(ds_name)
+            ]
+        else:
+            filtered_names = self.keys()
+
+        # Apply type filter if specified
+        if type_regex:
+            pattern = _compile_pattern(type_regex, type_regex_flags)
+            filtered_types = []
+            for ds_name in filtered_names:
+                # Retrieve the dataset type
+                if ds_name in self._lazy_datasets:
+                    str_type = str(self._lazy_datasets[ds_name])
+                else:
+                    class_type = type(self.__datasets[ds_name])
+                    str_type = f"{class_type.__module__}.{class_type.__qualname__}"
+                # Match the dataset type against the type_regex
+                if pattern.search(str_type):
+                    filtered_types.append(ds_name)
+
+            return filtered_types
+
+        return filtered_names
+
     def list(
         self, regex_search: str | None = None, regex_flags: int | re.RegexFlag = 0
     ) -> List[str]:  # noqa: UP006
-        # TODO: rename depending on the solution for https://github.com/kedro-org/kedro/issues/3917
-        # TODO: make regex_search mandatory argument as we have catalog.keys() for listing all the datasets.
+        # TODO: remove when removing old catalog
         """List all dataset names registered in the catalog, optionally filtered by a regex pattern.
 
         If a regex pattern is provided, only dataset names matching the pattern will be returned.
@@ -590,12 +663,7 @@ def list(
         if not regex_flags:
             regex_flags = re.IGNORECASE
 
-        try:
-            pattern = re.compile(regex_search, flags=regex_flags)
-        except re.error as exc:
-            raise SyntaxError(
-                f"Invalid regular expression provided: '{regex_search}'"
-            ) from exc
+        pattern = _compile_pattern(regex_search, regex_flags)
         return [ds_name for ds_name in self.__iter__() if pattern.search(ds_name)]
 
     def save(self, name: str, data: Any) -> None:
@@ -745,3 +813,12 @@ def exists(self, name: str) -> bool:
         except DatasetNotFoundError:
             return False
         return dataset.exists()
+
+
+def _compile_pattern(regex: str, regex_flags: int | re.RegexFlag) -> re.Pattern:
+    try:
+        pattern = re.compile(regex, flags=regex_flags)
+    except re.error as exc:
+        raise SyntaxError(f"Invalid regular expression provided: '{regex}'") from exc
+
+    return pattern

tests/io/test_kedro_data_catalog.py

@@ -47,9 +47,10 @@ def conflicting_feed_dict():
 
 @pytest.fixture
 def multi_catalog():
-    csv = CSVDataset(filepath="abc.csv")
+    csv_1 = CSVDataset(filepath="abc.csv")
+    csv_2 = CSVDataset(filepath="def.csv")
     parq = ParquetDataset(filepath="xyz.parq")
-    return KedroDataCatalog({"abc": csv, "xyz": parq})
+    return KedroDataCatalog({"abc": csv_1, "def": csv_2, "xyz": parq})
 
 
 @pytest.fixture
@@ -159,8 +160,9 @@ def test_multi_catalog_list(self, multi_catalog):
         [
             ("^a", ["abc"]),
             ("a|x", ["abc", "xyz"]),
-            ("^(?!(a|x))", []),
-            ("def", []),
+            ("^(?!(a|d|x))", []),
+            ("def", ["def"]),
+            ("ghi", []),
             ("", []),
         ],
     )
@@ -175,6 +177,61 @@ def test_multi_catalog_list_bad_regex(self, multi_catalog):
         with pytest.raises(SyntaxError, match=pattern):
             multi_catalog.list("((")
 
+    @pytest.mark.parametrize(
+        "name_regex,type_regex,expected",
+        [
+            ("^a", None, ["abc"]),
+            ("a|x", None, ["abc", "xyz"]),
+            ("a|d|x", None, ["abc", "def", "xyz"]),
+            ("a|d|x", "CSVDataset", ["abc", "def"]),
+            ("a|d|x", "kedro_datasets", ["abc", "def", "xyz"]),
+            (None, "ParquetDataset", ["xyz"]),
+            ("^(?!(a|d|x))", None, []),
+            ("def", None, ["def"]),
+            (None, None, ["abc", "def", "xyz"]),
+            ("a|d|x", "no_such_dataset", []),
+        ],
+    )
+    def test_catalog_filter_regex(
+        self, multi_catalog, name_regex, type_regex, expected
+    ):
+        """Test that regex patterns filter materialized datasets accordingly"""
+        assert (
+            multi_catalog.filter(name_regex=name_regex, type_regex=type_regex)
+            == expected
+        )
+
+    @pytest.mark.parametrize(
+        "name_regex,type_regex,expected",
+        [
+            ("b|m", None, ["boats", "materialized"]),
+            (None, None, ["boats", "cars", "materialized"]),
+            (None, "CSVDataset", ["boats", "cars"]),
+            (None, "ParquetDataset", ["materialized"]),
+            ("b|c", "ParquetDataset", []),
+        ],
+    )
+    def test_from_config_catalog_filter_regex(
+        self, data_catalog_from_config, name_regex, type_regex, expected
+    ):
+        """Test that regex patterns filter lazy and materialized datasets accordingly"""
+        data_catalog_from_config["materialized"] = ParquetDataset(filepath="xyz.parq")
+        assert (
+            data_catalog_from_config.filter(
+                name_regex=name_regex, type_regex=type_regex
+            )
+            == expected
+        )
+
+    def test_catalog_filter_bad_regex(self, multi_catalog):
+        """Test that bad regex is caught accordingly"""
+        escaped_regex = r"\(\("
+        pattern = f"Invalid regular expression provided: '{escaped_regex}'"
+        with pytest.raises(SyntaxError, match=pattern):
+            multi_catalog.filter(name_regex="((")
+        with pytest.raises(SyntaxError, match=pattern):
+            multi_catalog.filter(type_regex="((")
+
     def test_eq(self, multi_catalog, data_catalog):
         assert multi_catalog == multi_catalog.shallow_copy()
         assert multi_catalog != data_catalog
@@ -266,12 +323,14 @@ def test_init_with_raw_data(self, dummy_dataframe, dataset):
         assert isinstance(catalog["ds"], CSVDataset)
         assert isinstance(catalog["df"], MemoryDataset)
 
-    def test_repr(self, data_catalog):
-        assert data_catalog.__repr__() == str(data_catalog)
+    def test_repr(self, data_catalog_from_config):
+        assert data_catalog_from_config.__repr__() == str(data_catalog_from_config)
 
     def test_repr_no_type_found(self, data_catalog_from_config):
         del data_catalog_from_config._lazy_datasets["boats"].config["type"]
-        assert data_catalog_from_config.__repr__() == str(data_catalog_from_config)
+        pattern = "'type' is missing from dataset catalog configuration"
+        with pytest.raises(DatasetError, match=re.escape(pattern)):
+            _ = str(data_catalog_from_config)
 
     def test_missing_keys_from_load_versions(self, correct_config):
         """Test load versions include keys missing in the catalog"""