Remove cachetools dependency (#5352)

Cachetools 7 introduced several breaking changes that made @cachedmethod significantly stricter, including converting it into a descriptor and requiring instances to have a mutable __dict__.
These assumptions do not align well with Kedro’s ParallelRunner, which relies on pickling and unpickling dataset instances across processes, and this resulted in runtime failures when upgrading to cachetools 7. Although pinning the dependency avoided the immediate issue, further investigation showed that cachetools is not well-suited to this use case. Kedro only caches two values per dataset instance (the resolved load and save versions) and does not require eviction policies, TTLs, or multi-key caching.

As a long-term solution, this change removes the cachetools dependency and replaces it with lightweight manual caching. This simplifies the implementation, removes an unnecessary dependency, and avoids future breakages in parallel execution.

Author: rashidakanchwala

.github/workflows/e2e-tests.yml

@@ -33,6 +33,8 @@ jobs:
           activate-environment: true
       - name: Install dependencies
         run: |
+          # TEMP: For passing tests, will be removed before merge
+          uv pip install "git+https://github.com/kedro-org/kedro-plugins.git@fix/remove-cachetools-deps#subdirectory=kedro-datasets"
           make install-test-requirements
           uv pip install pip
           make install-pre-commit

.github/workflows/unit-tests.yml

@@ -30,6 +30,8 @@ jobs:
           activate-environment: true
       - name: Install dependencies
         run: |
+            # TEMP: For passing tests, will be removed before merge
+            uv pip install "git+https://github.com/kedro-org/kedro-plugins.git@fix/remove-cachetools-deps#subdirectory=kedro-datasets"
             make install-test-requirements
             make install-pre-commit
       - name: pip freeze

kedro/io/core.py

@@ -12,10 +12,9 @@
 import warnings
 from collections import namedtuple
 from datetime import datetime, timezone
-from functools import partial, wraps
+from functools import wraps
 from glob import iglob
 from inspect import getcallargs
-from operator import attrgetter
 from pathlib import Path, PurePath, PurePosixPath
 from typing import (  # noqa: UP035
     TYPE_CHECKING,
@@ -30,8 +29,6 @@
     runtime_checkable,
 )
 
-from cachetools import Cache, cachedmethod
-from cachetools.keys import hashkey
 from typing_extensions import Self
 
 # These are re-exported for backward compatibility
@@ -734,13 +731,23 @@ def __init__(
         self._version = version
         self._exists_function = exists_function or _local_exists
         self._glob_function = glob_function or iglob
-        # 1 entry for load version, 1 for save version
-        self._version_cache = Cache(maxsize=2)  # type: Cache
+        self._cached_load_version: str | None = None
+        self._cached_save_version: str | None = None
+
+    def _clear_version_cache(self) -> None:
+        """Clear both load and save version caches."""
+        self._cached_load_version = None
+        self._cached_save_version = None
 
-    # 'key' is set to prevent cache key overlapping for load and save:
-    # https://cachetools.readthedocs.io/en/stable/#cachetools.cachedmethod
-    @cachedmethod(cache=attrgetter("_version_cache"), key=partial(hashkey, "load"))
     def _fetch_latest_load_version(self) -> str:
+        """Fetch the most recent existing version from the given path.
+        Results are cached to avoid repeated filesystem operations.
+        """
+
+        # Return cached version if available
+        if self._cached_load_version is not None:
+            return self._cached_load_version
+
         # When load version is unpinned, fetch the most recent existing
         # version from the given path.
         pattern = str(self._get_versioned_path("*"))
@@ -758,14 +765,20 @@ def _fetch_latest_load_version(self) -> str:
         if not most_recent:
             message = f"Did not find any versions for {self}"
             raise VersionNotFoundError(message)
-        return PurePath(most_recent).parent.name
 
-    # 'key' is set to prevent cache key overlapping for load and save:
-    # https://cachetools.readthedocs.io/en/stable/#cachetools.cachedmethod
-    @cachedmethod(cache=attrgetter("_version_cache"), key=partial(hashkey, "save"))
+        # Cache and return the result
+        self._cached_load_version = PurePath(most_recent).parent.name
+        return self._cached_load_version
+
     def _fetch_latest_save_version(self) -> str:
         """Generate and cache the current save version"""
-        return generate_timestamp()
+        # Return cached version if available
+        if self._cached_save_version is not None:
+            return self._cached_save_version
+
+        # Generate new timestamp and cache it
+        self._cached_save_version = generate_timestamp()
+        return self._cached_save_version
 
     def resolve_load_version(self) -> str | None:
         """Compute the version the dataset should be loaded with."""
@@ -818,7 +831,7 @@ def _save_wrapper(
 
         @wraps(save_func)
         def save(self: Self, data: _DI) -> None:
-            self._version_cache.clear()
+            self._clear_version_cache()
             save_version = (
                 self.resolve_save_version()
             )  # Make sure last save version is set
@@ -844,7 +857,7 @@ def save(self: Self, data: _DI) -> None:
                 warnings.warn(
                     _CONSISTENCY_WARNING.format(save_version, load_version, str(self))
                 )
-                self._version_cache.clear()
+                self._clear_version_cache()
 
         return save
 
@@ -870,7 +883,7 @@ def exists(self) -> bool:
 
     def _release(self) -> None:
         super()._release()
-        self._version_cache.clear()
+        self._clear_version_cache()
 
 
 def get_protocol_and_path(

pyproject.toml

@@ -14,7 +14,6 @@ requires-python = ">=3.10"
 dependencies = [
     "attrs>=21.3",
     "build>=0.7.0",
-    "cachetools>=4.1,<7.0", # Pinned because cachetools 7 breaks Kedro multiprocessing
     "click>=8.2",
     "cookiecutter>=2.1.1,<3.0",
     "dynaconf>=3.1.2,<4.0",
@@ -73,7 +72,6 @@ test = [
     # mypy related dependencies
     "pandas-stubs",
     "types-PyYAML",
-    "types-cachetools",
     "types-requests",
 ]
 docs = [

tests/io/test_core.py

@@ -598,11 +598,28 @@ def test_versioning_existing_dataset(
         assert my_versioned_dataset.exists()
 
     def test_cache_release(self, my_versioned_dataset):
-        my_versioned_dataset._version_cache["index"] = "value"
-        assert my_versioned_dataset._version_cache.currsize > 0
+        # Set cache values
+        my_versioned_dataset._cached_load_version = "2024-01-01T00.00.00.000Z"
+        my_versioned_dataset._cached_save_version = "2024-01-02T00.00.00.000Z"
+        assert my_versioned_dataset._cached_load_version is not None
+        assert my_versioned_dataset._cached_save_version is not None
 
+        # Release should clear the cache
         my_versioned_dataset._release()
-        assert my_versioned_dataset._version_cache.currsize == 0
+        assert my_versioned_dataset._cached_load_version is None
+        assert my_versioned_dataset._cached_save_version is None
+
+    def test_clear_version_cache(self, my_versioned_dataset):
+        # Set cache values
+        my_versioned_dataset._cached_load_version = "2024-01-01T00.00.00.000Z"
+        my_versioned_dataset._cached_save_version = "2024-01-02T00.00.00.000Z"
+        assert my_versioned_dataset._cached_load_version is not None
+        assert my_versioned_dataset._cached_save_version is not None
+
+        # Clear cache should reset both to None
+        my_versioned_dataset._clear_version_cache()
+        assert my_versioned_dataset._cached_load_version is None
+        assert my_versioned_dataset._cached_save_version is None
 
     def test_fetch_latest_load_version_success(self, my_versioned_dataset, mocker):
         mock_get_versioned_path = mocker.patch(
@@ -622,7 +639,7 @@ def test_fetch_latest_load_version_success(self, my_versioned_dataset, mocker):
         )
 
         # Ensure the cache is empty for the test
-        my_versioned_dataset._version_cache = {}
+        my_versioned_dataset._cached_load_version = None
         result = my_versioned_dataset._fetch_latest_load_version()
 
         mock_get_versioned_path.assert_called_once_with("*")