Repository: joblib/joblib

Issue:
Expiry of old results
I've added a small extra feature that recalculates if a result is older than a certain age. 

```
@memory.cache(expires_after=60)
def f(x):
 ...
```

Will recompute if cached result is older than 60 seconds. Code is in this branch:
https://github.com/fredludlow/joblib/tree/expires-after

Is this of interest? And if so, is there anything else that needs doing before sending a pull request? (I've added test coverage in test_memory.py)


Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=h1) Report
> Merging [#1149](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=desc) (203d43f) into [master](https://codecov.io/gh/joblib/joblib/commit/c952c223e3616e5ff10f63864b4615088b0d9352?el=desc) (c952c22) will **decrease** coverage by `14.39%`.
> The diff coverage is `98.93%`.

[![Impacted file tree graph](https://codecov.io/gh/joblib/joblib/pull/1149/graphs/tree.svg?width=650&height=150&src=pr&token=gA6LF5DGTW)](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##           master    #1149       +/-   ##
===========================================
- Coverage   94.44%   80.04%   -14.40%     
===========================================
  Files          47       47               
  Lines        6964     6856      -108     
===========================================
- Hits         6577     5488     -1089     
- Misses        387     1368      +981     
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/memory.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL21lbW9yeS5weQ==) | `95.23% <96.77%> (-0.33%)` | :arrow_down: |
| [joblib/test/test\_memory.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1vcnkucHk=) | `98.49% <100.00%> (-0.08%)` | :arrow_down: |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `3.77% <0.00%> (-95.11%)` | :arrow_down: |
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL19kYXNrLnB5) | `23.76% <0.00%> (-70.33%)` | :arrow_down: |
| [joblib/test/test\_memmapping.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1tYXBwaW5nLnB5) | `37.07% <0.00%> (-62.17%)` | :arrow_down: |
| [joblib/format\_stack.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL2Zvcm1hdF9zdGFjay5weQ==) | `58.33% <0.00%> (-41.67%)` | :arrow_down: |
| [joblib/backports.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL2JhY2twb3J0cy5weQ==) | `44.73% <0.00%> (-39.48%)` | :arrow_down: |
| [joblib/my\_exceptions.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL215X2V4Y2VwdGlvbnMucHk=) | `64.28% <0.00%> (-35.72%)` | :arrow_down: |
| [joblib/\_memmapping\_reducer.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL19tZW1tYXBwaW5nX3JlZHVjZXIucHk=) | `61.74% <0.00%> (-34.87%)` | :arrow_down: |
| [joblib/executor.py](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree#diff-am9ibGliL2V4ZWN1dG9yLnB5) | `76.19% <0.00%> (-23.81%)` | :arrow_down: |
| ... and [23 more](https://codecov.io/gh/joblib/joblib/pull/1149/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=footer). Last update [c952c22...203d43f](https://codecov.io/gh/joblib/joblib/pull/1149?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

jjerphan: My previous review might be outdated, let me know if you want another one, @tomMoral.
xloem: This popular feature appears to have rested unmerged after passing an extended review. Sadly there are now conflicts and test failures again. Is this good to merge if it is ported forward?
tomMoral: I just fixed this, I will try to merge it to include it in the next release.

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index b6c3610..25d3f27 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -32,6 +32,10 @@ In development
   previous versions of Joblib.
   https://github.com/joblib/joblib/pull/1374
 
+- Add ``cache_validation_callback`` in :meth:`joblib.Memory.cache`, to allow
+  custom cache invalidation based on the metadata of the function call.
+  https://github.com/joblib/joblib/pull/1149
+
 - Add a ``return_generator`` parameter for ``Parallel``, that allows
   to consume results asynchronously.
   https://github.com/joblib/joblib/pull/1393
diff --git a/doc/memory.rst b/doc/memory.rst
index 7b6b176..0a6abd8 100644
--- a/doc/memory.rst
+++ b/doc/memory.rst
@@ -375,9 +375,9 @@ Gotchas
      ``self.method = memory.cache(self.method, ignore=['self'])``.
 
 * **joblib cache entries may be invalidated after environment updates**.
-  Values returned by ``joblib.hash`` are not guaranteed to stay
+  Values returned by :func:`joblib.hash` are not guaranteed to stay
   constant across ``joblib`` versions. This means that **all** entries of a
-  ``joblib.Memory`` cache can get invalidated when upgrading ``joblib``.
+  :class:`Memory` cache can get invalidated when upgrading ``joblib``.
   Invalidation can also happen when upgrading a third party library (such as
   ``numpy``): in such a case, only the cached function calls with parameters
   that are constructs (or contain references to constructs) defined in the
@@ -388,7 +388,8 @@ Ignoring some arguments
 -----------------------
 
 It may be useful not to recalculate a function when certain arguments
-change, for instance a debug flag. `Memory` provides the `ignore` list::
+change, for instance a debug flag. :class:`Memory` provides the ``ignore``
+list::
 
     >>> @memory.cache(ignore=['debug'])
     ... def my_func(x, debug=True):
@@ -400,6 +401,62 @@ change, for instance a debug flag. `Memory` provides the `ignore` list::
     >>> # my_func was not reevaluated
 
 
+Custom cache validation
+-----------------------
+
+In some cases, external factors can invalidate the cached results and
+one wants to have more control on whether to reuse a result or not.
+
+This is for instance the case if the results depends on database records
+that change over time: a small delay in the updates might be tolerable
+but after a while, the results might be invalid.
+
+One can have a finer control on the cache validity specifying a function
+via ``cache_validation_callback`` in :meth:`~joblib.Memory.cache`. For
+instance, one can only cache results that take more than 1s to be computed.
+
+    >>> import time
+    >>> def cache_validation_cb(metadata):
+    ...     # Only retrieve cached results for calls that take more than 1s
+    ...     return metadata['duration'] > 1
+
+    >>> @memory.cache(cache_validation_callback=cache_validation_cb)
+    ... def my_func(delay=0):
+    ...     time.sleep(delay)
+    ...	    print(f'Called with {delay}s delay')
+
+    >>> my_func()
+    Called with 0s delay
+    >>> my_func(1.1)
+    Called with 1.1s delay
+    >>> my_func(1.1)  # This result is retrieved from cache
+    >>> my_func()  # This one is not and the call is repeated
+    Called with 0s delay
+
+``cache_validation_cb`` will be called with a single argument containing
+the metadata of the cached call as a dictionary containing the following
+keys:
+
+  - ``duration``: the duration of the function call,
+  - ``time``: the timestamp when the cache called has been recorded
+  - ``input_args``: a dictionary of keywords arguments for the cached function call.
+
+Note a validity duration for cached results can be defined via
+:func:`joblib.expires_after` by providing similar with arguments similar to the
+ones of a ``datetime.timedelta``:
+
+    >>> from joblib import expires_after
+    >>> @memory.cache(cache_validation_callback=expires_after(seconds=0.5))
+    ... def my_func():
+    ...	    print(f'Function run')
+    >>> my_func()
+    Function run
+    >>> my_func()
+    >>> time.sleep(0.5)
+    >>> my_func()
+    Function run
+
+
 .. _memory_reference:
 
 Reference documentation of the :class:`~joblib.Memory` class
@@ -448,3 +505,9 @@ without actually needing to call the function itself::
     ...     shutil.rmtree(cachedir2)
     ... except OSError:
     ...     pass  # this can sometimes fail under Windows
+
+
+Helper Reference
+~~~~~~~~~~~~~~~~
+
+.. autofunction:: joblib.expires_after
diff --git a/examples/serialization_and_wrappers.py b/examples/serialization_and_wrappers.py
index d463242..0b21564 100644
--- a/examples/serialization_and_wrappers.py
+++ b/examples/serialization_and_wrappers.py
@@ -118,7 +118,7 @@ except Exception:
 ###############################################################################
 # To have both fast pickling, safe process creation and serialization of
 # interactive functions, ``loky`` provides a wrapper function
-# :func:`wrap_non_picklable_objects` to wrap the non-picklable function and
+# ``wrap_non_picklable_objects`` to wrap the non-picklable function and
 # indicate to the serialization process that this specific function should be
 # serialized using ``cloudpickle``. This changes the serialization behavior
 # only for this function and keeps using ``pickle`` for all other objects. The
diff --git a/joblib/__init__.py b/joblib/__init__.py
index bc34d81..b5be5fe 100644
--- a/joblib/__init__.py
+++ b/joblib/__init__.py
@@ -110,13 +110,22 @@ __version__ = '1.3.0.dev0'
 
 
 import os
-from .memory import Memory, MemorizedResult, register_store_backend
+
+from .memory import Memory
+from .memory import MemorizedResult
+from .memory import register_store_backend
+from .memory import expires_after
+
 from .logger import PrintTime
 from .logger import Logger
+
 from .hashing import hash
+
 from .numpy_pickle import dump
 from .numpy_pickle import load
+
 from .compressor import register_compressor
+
 from .parallel import Parallel
 from .parallel import delayed
 from .parallel import cpu_count
@@ -129,7 +138,7 @@ from ._cloudpickle_wrapper import wrap_non_picklable_objects
 
 __all__ = ['Memory', 'MemorizedResult', 'PrintTime', 'Logger', 'hash', 'dump',
            'load', 'Parallel', 'delayed', 'cpu_count', 'effective_n_jobs',
-           'register_parallel_backend', 'parallel_backend',
+           'register_parallel_backend', 'parallel_backend', 'expires_after',
            'register_store_backend', 'register_compressor',
            'wrap_non_picklable_objects', 'parallel_config']
 
diff --git a/joblib/memory.py b/joblib/memory.py
index cc832c8..8171d24 100644
--- a/joblib/memory.py
+++ b/joblib/memory.py
@@ -22,6 +22,7 @@ import traceback
 import warnings
 import inspect
 import weakref
+from datetime import timedelta
 
 from tokenize import open as open_py_source
 
@@ -408,17 +409,26 @@ class MemorizedFunc(Logger):
     verbose: int, optional
         The verbosity flag, controls messages that are issued as
         the function is evaluated.
+
+    cache_validation_callback: callable, optional
+        Callable to check if a result in cache is valid or is to be recomputed.
+        When the function is called with arguments for which a cache exists,
+        the callback is called with the cache entry's metadata as its sole
+        argument. If it returns True, the cached result is returned, else the
+        cache for these arguments is cleared and the result is recomputed.
     """
     # ------------------------------------------------------------------------
     # Public interface
     # ------------------------------------------------------------------------
 
     def __init__(self, func, location, backend='local', ignore=None,
-                 mmap_mode=None, compress=False, verbose=1, timestamp=None):
+                 mmap_mode=None, compress=False, verbose=1, timestamp=None,
+                 cache_validation_callback=None):
         Logger.__init__(self)
         self.mmap_mode = mmap_mode
         self.compress = compress
         self.func = func
+        self.cache_validation_callback = cache_validation_callback
 
         if ignore is None:
             ignore = []
@@ -434,15 +444,16 @@ class MemorizedFunc(Logger):
                                                     )
         if self.store_backend is not None:
             # Create func directory on demand.
-            self.store_backend.\
-                store_cached_func_code([_build_func_identifier(self.func)])
+            self.store_backend.store_cached_func_code([
+                _build_func_identifier(self.func)
+            ])
 
         if timestamp is None:
             timestamp = time.time()
         self.timestamp = timestamp
         try:
             functools.update_wrapper(self, func)
-        except:  # noqa: E722
+        except Exception:
             " Objects like ufunc don't like that "
         if inspect.isfunction(func):
             doc = pydoc.TextDoc().document(func)
@@ -458,6 +469,34 @@ class MemorizedFunc(Logger):
         self._func_code_info = None
         self._func_code_id = None
 
+    def _is_in_cache_and_valid(self, path):
+        """Check if the function call is cached and valid for given arguments.
+
+        - Compare the function code with the one from the cached function,
+        asserting if it has changed.
+        - Check if the function call is present in the cache.
+        - Call `cache_validation_callback` for user define cache validation.
+
+        Returns True if the function call is in cache and can be used, and
+        returns False otherwise.
+        """
+        # Check if the code of the function has changed
+        if not self._check_previous_func_code(stacklevel=4):
+            return False
+
+        # Check if this specific call is in the cache
+        if not self.store_backend.contains_item(path):
+            return False
+
+        # Call the user defined cache validation callback
+        metadata = self.store_backend.get_metadata(path)
+        if (self.cache_validation_callback is not None and
+                not self.cache_validation_callback(metadata)):
+            self.store_backend.clear_item(path)
+            return False
+
+        return True
+
     def _cached_call(self, args, kwargs, shelving=False):
         """Call wrapped function and cache result, or read cache if available.
 
@@ -513,20 +552,10 @@ class MemorizedFunc(Logger):
                 )
             )
 
-        # FIXME: The statements below should be try/excepted
         # Compare the function code with the previous to see if the
-        # function code has changed
-        if not (self._check_previous_func_code(stacklevel=4) and
-                self.store_backend.contains_item([func_id, args_id])):
-            if self._verbose > 10:
-                _, name = get_func_name(self.func)
-                self.warn('Computing func {0}, argument hash {1} '
-                          'in location {2}'
-                          .format(name, args_id,
-                                  self.store_backend.
-                                  get_cached_func_info([func_id])['location']))
-            must_call = True
-        else:
+        # function code has changed and check if the results are present in
+        # the cache.
+        if self._is_in_cache_and_valid([func_id, args_id]):
             try:
                 t0 = time.time()
                 if self._verbose:
@@ -555,6 +584,15 @@ class MemorizedFunc(Logger):
                           '{}\n {}'.format(signature, traceback.format_exc()))
 
                 must_call = True
+        else:
+            if self._verbose > 10:
+                _, name = get_func_name(self.func)
+                self.warn('Computing func {0}, argument hash {1} '
+                          'in location {2}'
+                          .format(name, args_id,
+                                  self.store_backend.
+                                  get_cached_func_info([func_id])['location']))
+            must_call = True
 
         if must_call:
             out, metadata = self.call(*args, **kwargs)
@@ -852,7 +890,9 @@ class MemorizedFunc(Logger):
         input_repr = dict((k, repr(v)) for k, v in argument_dict.items())
         # This can fail due to race-conditions with multiple
         # concurrent joblibs removing the file or the directory
-        metadata = {"duration": duration, "input_args": input_repr}
+        metadata = {
+            "duration": duration, "input_args": input_repr, "time": start_time,
+        }
 
         func_id, args_id = self._get_output_identifiers(*args, **kwargs)
         self.store_backend.store_metadata([func_id, args_id], metadata)
@@ -982,7 +1022,8 @@ class Memory(Logger):
             backend_options=dict(compress=compress, mmap_mode=mmap_mode,
                                  **backend_options))
 
-    def cache(self, func=None, ignore=None, verbose=None, mmap_mode=False):
+    def cache(self, func=None, ignore=None, verbose=None, mmap_mode=False,
+              cache_validation_callback=None):
         """ Decorates the given function func to only compute its return
             value for input arguments not cached on disk.
 
@@ -999,6 +1040,13 @@ class Memory(Logger):
                 The memmapping mode used when loading from cache
                 numpy arrays. See numpy.load for the meaning of the
                 arguments. By default that of the memory object is used.
+            cache_validation_callback: callable, optional
+                Callable to validate whether or not the cache is valid. When
+                the cached function is called with arguments for which a cache
+                exists, this callable is called with the metadata of the cached
+                result as its sole argument. If it returns True, then the
+                cached result is returned, else the cache for these arguments
+                is cleared and recomputed.
 
             Returns
             -------
@@ -1008,11 +1056,21 @@ class Memory(Logger):
                 methods for cache lookup and management. See the
                 documentation for :class:`joblib.memory.MemorizedFunc`.
         """
+        if (cache_validation_callback is not None and
+                not callable(cache_validation_callback)):
+            raise ValueError(
+                "cache_validation_callback needs to be callable. "
+                f"Got {cache_validation_callback}."
+            )
         if func is None:
             # Partial application, to be able to specify extra keyword
             # arguments in decorators
-            return functools.partial(self.cache, ignore=ignore,
-                                     verbose=verbose, mmap_mode=mmap_mode)
+            return functools.partial(
+                self.cache, ignore=ignore,
+                mmap_mode=mmap_mode,
+                verbose=verbose,
+                cache_validation_callback=cache_validation_callback
+            )
         if self.store_backend is None:
             return NotMemorizedFunc(func)
         if verbose is None:
@@ -1021,11 +1079,12 @@ class Memory(Logger):
             mmap_mode = self.mmap_mode
         if isinstance(func, MemorizedFunc):
             func = func.func
-        return MemorizedFunc(func, location=self.store_backend,
-                             backend=self.backend,
-                             ignore=ignore, mmap_mode=mmap_mode,
-                             compress=self.compress,
-                             verbose=verbose, timestamp=self.timestamp)
+        return MemorizedFunc(
+            func, location=self.store_backend, backend=self.backend,
+            ignore=ignore, mmap_mode=mmap_mode, compress=self.compress,
+            verbose=verbose, timestamp=self.timestamp,
+            cache_validation_callback=cache_validation_callback
+        )
 
     def clear(self, warn=True):
         """ Erase the complete cache directory.
@@ -1113,3 +1172,28 @@ class Memory(Logger):
         state = self.__dict__.copy()
         state['timestamp'] = None
         return state
+
+
+###############################################################################
+# cache_validation_callback helpers
+###############################################################################
+
+def expires_after(days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0,
+                  hours=0, weeks=0):
+    """Helper cache_validation_callback to force recompute after a duration.
+
+    Parameters
+    ----------
+    days, seconds, microseconds, milliseconds, minutes, hours, weeks: numbers
+        argument passed to a timedelta.
+    """
+    delta = timedelta(
+        days=days, seconds=seconds, microseconds=microseconds,
+        milliseconds=milliseconds, minutes=minutes, hours=hours, weeks=weeks
+    )
+
+    def cache_validation_callback(metadata):
+        computation_age = time.time() - metadata['time']
+        return computation_age < delta.total_seconds()
+
+    return cache_validation_callback


Test patch:
diff --git a/joblib/test/test_memory.py b/joblib/test/test_memory.py
index 98cfce6..ff018fb 100644
--- a/joblib/test/test_memory.py
+++ b/joblib/test/test_memory.py
@@ -22,6 +22,7 @@ import textwrap
 import pytest
 
 from joblib.memory import Memory
+from joblib.memory import expires_after
 from joblib.memory import MemorizedFunc, NotMemorizedFunc
 from joblib.memory import MemorizedResult, NotMemorizedResult
 from joblib.memory import _FUNCTION_HASHES
@@ -87,10 +88,10 @@ def test_memory_integration(tmpdir):
     """ Simple test of memory lazy evaluation.
     """
     accumulator = list()
+
     # Rmk: this function has the same name than a module-level function,
     # thus it serves as a test to see that both are identified
     # as different.
-
     def f(arg):
         accumulator.append(1)
         return arg
@@ -1401,3 +1402,87 @@ def test_deprecated_bytes_limit(tmpdir):
         )
     with pytest.warns(DeprecationWarning, match="bytes_limit"):
         _ = Memory(location=tmpdir.strpath, bytes_limit='1K')
+
+
+class TestCacheValidationCallback:
+    "Tests on parameter `cache_validation_callback`"
+
+    @pytest.fixture()
+    def memory(self, tmp_path):
+        mem = Memory(location=tmp_path)
+        yield mem
+        mem.clear()
+
+    def foo(self, x, d, delay=None):
+        d["run"] = True
+        if delay is not None:
+            time.sleep(delay)
+        return x * 2
+
+    def test_invalid_cache_validation_callback(self, memory):
+        "Test invalid values for `cache_validation_callback"
+        match = "cache_validation_callback needs to be callable. Got True."
+        with pytest.raises(ValueError, match=match):
+            memory.cache(cache_validation_callback=True)
+
+    @pytest.mark.parametrize("consider_cache_valid", [True, False])
+    def test_constant_cache_validation_callback(
+            self, memory, consider_cache_valid
+    ):
+        "Test expiry of old results"
+        f = memory.cache(
+            self.foo, cache_validation_callback=lambda _: consider_cache_valid,
+            ignore=["d"]
+        )
+
+        d1, d2 = {"run": False}, {"run": False}
+        assert f(2, d1) == 4
+        assert f(2, d2) == 4
+
+        assert d1["run"]
+        assert d2["run"] != consider_cache_valid
+
+    def test_memory_only_cache_long_run(self, memory):
+        "Test cache validity based on run duration."
+
+        def cache_validation_callback(metadata):
+            duration = metadata['duration']
+            if duration > 0.1:
+                return True
+
+        f = memory.cache(
+            self.foo, cache_validation_callback=cache_validation_callback,
+            ignore=["d"]
+        )
+
+        # Short run are not cached
+        d1, d2 = {"run": False}, {"run": False}
+        assert f(2, d1, delay=0) == 4
+        assert f(2, d2, delay=0) == 4
+        assert d1["run"]
+        assert d2["run"]
+
+        # Longer run are cached
+        d1, d2 = {"run": False}, {"run": False}
+        assert f(2, d1, delay=0.2) == 4
+        assert f(2, d2, delay=0.2) == 4
+        assert d1["run"]
+        assert not d2["run"]
+
+    def test_memory_expires_after(self, memory):
+        "Test expiry of old cached results"
+
+        f = memory.cache(
+            self.foo, cache_validation_callback=expires_after(seconds=.3),
+            ignore=["d"]
+        )
+
+        d1, d2, d3 = {"run": False}, {"run": False}, {"run": False}
+        assert f(2, d1) == 4
+        assert f(2, d2) == 4
+        time.sleep(.5)
+        assert f(2, d3) == 4
+
+        assert d1["run"]
+        assert not d2["run"]
+        assert d3["run"]