Callable timeout

CHANGES.rst

@@ -2,6 +2,12 @@ Changelog
 =========
 
 
+Version 2.4.0
+-------------
+
+- added support for callable `timeout` in `@cached` and `@memoized`
+
+
 Version 2.3.0
 -------------
 
@@ -28,6 +34,7 @@ Released 2024-10-08
 - support Flask 3
 
 
+
 Version 2.0.2
 -------------
 
@@ -38,6 +45,7 @@ Released 2023-01-12
 - bug fix: make the ``make_cache_key`` attributed of decorated view functions writeable. :pr:`431`, :issue:`97`
 
 
+
 Version 2.0.1
 -------------
 

docs/index.rst

@@ -87,6 +87,11 @@ a subclass of `flask.Response`::
             timeout=50,
         )
 
+.. versionchanged:: 2.0.3
+    A dynamic timeout can also be achieved via setting ``@cached``'s `timeout` argument
+    to a callable which takes the decorated function's output as a positional argument
+    and returns `None` or an integer. Callable timeout is also available to ``@memoized``.
+
 .. warning::
 
     When using ``cached`` on a view, take care to put it between Flask's
@@ -206,6 +211,11 @@ every time this information is needed you might do something like the following:
             def __repr__(self):
                 return "%s(%s)" % (self.__class__.__name__, self.id)
 
+.. versionchanged:: 2.0.3
+    A dynamic timeout can be achieved via setting ``@memoized``'s `timeout` argument
+    to a callable which takes the decorated function's output as a positional argument
+    and returns `None` or an integer. Callable timeout is also available to ``@cached``.
+
 
 Deleting memoize cache
 ``````````````````````

src/flask_caching/__init__.py

@@ -241,7 +241,7 @@ def unlink(self, *args, **kwargs) -> List[str]:
 
     def cached(
         self,
-        timeout: Optional[int] = None,
+        timeout: Optional[Union[int, Callable]] = None,
         key_prefix: str = "view/%s",
         unless: Optional[Callable] = None,
         forced_update: Optional[Callable] = None,
@@ -297,6 +297,11 @@ def get_list():
         :param timeout: Default None. If set to an integer, will cache for that
                         amount of time. Unit of time is in seconds.
 
+                        .. versionchanged:: 2.0.3
+                            Can optionally be a callable which expects one
+                            argument, the result of the cached function
+                            evaluation, and returns None or an integer.
+
         :param key_prefix: Default 'view/%(request.path)s'. Beginning key to .
                            use for the cache key. `request.path` will be the
                            actual request path, or in cases where the
@@ -428,14 +433,17 @@ def apply_caching(response):
                         rv = [val for val in rv]
 
                     if response_filter is None or response_filter(rv):
-                        cache_timeout = decorated_function.cache_timeout
+                        timeout = decorated_function.cache_timeout
                         if isinstance(rv, CachedResponse):
-                            cache_timeout = rv.timeout or cache_timeout
+                            timeout = rv.timeout or timeout
+                        elif callable(timeout):
+                            timeout = timeout(rv)
+
                         try:
                             self.cache.set(
                                 cache_key,
                                 rv,
-                                timeout=cache_timeout,
+                                timeout=timeout,
                             )
                         except Exception:
                             if self.app.debug:
@@ -615,6 +623,9 @@ def _memoize_make_cache_key(
 
         def make_cache_key(f, *args, **kwargs):
             _timeout = getattr(timeout, "cache_timeout", timeout)
+            if callable(_timeout):
+                _timeout = 0  # placeholder until timeout(rv) is doable
+
             fname, version_data = self._memoize_version(
                 f,
                 args=args,
@@ -747,7 +758,7 @@ def _bypass_cache(
 
     def memoize(
         self,
-        timeout: Optional[int] = None,
+        timeout: Optional[Union[int, Callable]] = None,
         make_name: Optional[Callable] = None,
         unless: Optional[Callable] = None,
         forced_update: Optional[Callable] = None,
@@ -800,17 +811,26 @@ def big_foo(a, b):
 
         :param timeout: Default None. If set to an integer, will cache for that
                         amount of time. Unit of time is in seconds.
+
+                        .. versionchanged:: 2.0.3
+                            Can optionally be a callable which expects one
+                            argument, the result of the cached function
+                            evaluation, and returns None or an integer.
+
         :param make_name: Default None. If set this is a function that accepts
                           a single argument, the function name, and returns a
                           new string to be used as the function name.
                           If not set then the function name is used.
+
         :param unless: Default None. Cache will *always* execute the caching
                        facilities unless this callable is true.
                        This will bypass the caching entirely.
+
         :param forced_update: Default None. If this callable is true,
                               cache value will be updated regardless cache
                               is expired or not. Useful for background
                               renewal of cached functions.
+
         :param response_filter: Default None. If not None, the callable is
                                 invoked after the cached funtion evaluation,
                                 and is given one arguement, the response
@@ -819,6 +839,7 @@ def big_foo(a, b):
                                 caching of code 500 responses.
         :param hash_method: Default hashlib.md5. The hash method used to
                             generate the keys for cached results.
+
         :param cache_none: Default False. If set to True, add a key exists
                            check when cache.get returns None. This will likely
                            lead to wrongly returned None values in concurrent
@@ -833,6 +854,7 @@ def big_foo(a, b):
                              formed with the function's source code hash in
                              addition to other parameters that may be included
                              in the formation of the key.
+
         :param args_to_ignore: List of arguments that will be ignored while
                                generating the cache key. Default to None.
                                This means that those arguments may change
@@ -901,11 +923,15 @@ def decorated_function(*args, **kwargs):
                         rv = [val for val in rv]
 
                     if response_filter is None or response_filter(rv):
+                        timeout = decorated_function.cache_timeout
+                        if callable(timeout):
+                            timeout = timeout(rv)
+
                         try:
                             self.cache.set(
                                 cache_key,
                                 rv,
-                                timeout=decorated_function.cache_timeout,
+                                timeout=timeout,
                             )
                         except Exception:
                             if self.app.debug:

tests/conftest.py

@@ -7,13 +7,15 @@
 import flask_caching as fsc
 
 try:
-    __import__("pytest_xprocess")
     from xprocess import ProcessStarter
 except ImportError:
+    try:
+        __import__("pytest_xprocess")
+    except ImportError:
 
-    @pytest.fixture(scope="session")
-    def xprocess():
-        pytest.skip("pytest-xprocess not installed.")
+        @pytest.fixture(scope="session")
+        def xprocess():
+            pytest.skip("pytest-xprocess not installed.")
 
 
 @pytest.fixture

tests/test_memoize.py

@@ -73,6 +73,31 @@ def big_foo(a, b):
         assert big_foo(5, 2) != result
 
 
+def test_memoize_dynamic_timeout_via_callable_timeout(app):
+    app.config["CACHE_DEFAULT_TIMEOUT"] = 1
+    cache = Cache(app)
+
+    with app.test_request_context():
+
+        @cache.memoize(
+            # This should override the timeout to be 2 seconds
+            timeout=lambda rv: 2
+            if isinstance(rv, int)
+            else 1
+        )
+        def big_foo(a, b):
+            return a + b + random.randrange(0, 100000)
+
+        result = big_foo(5, 2)
+        assert big_foo(5, 2) == result
+
+        time.sleep(1)  # after 1 second, cache is still active
+        assert big_foo(5, 2) == result
+
+        time.sleep(1)  # after 2 seconds, cache is not still active
+        assert big_foo(5, 2) != result
+
+
 def test_memoize_annotated(app, cache):
     with app.test_request_context():
 

tests/test_view.py

@@ -3,6 +3,7 @@
 
 from flask import make_response
 from flask import request
+from flask import Response
 from flask.views import View
 
 from flask_caching import CachedResponse
@@ -307,7 +308,7 @@ def cached_view2(foo, bar):
     assert time2 != tc.get("/a/b").data.decode("utf-8")
 
 
-def test_cache_timeout_dynamic(app, cache):
+def test_cache_timeout_dynamic_via_cached_reponse(app, cache):
     @app.route("/")
     @cache.cached(timeout=1)
     def cached_view():
@@ -327,6 +328,46 @@ def cached_view():
     assert time1 != tc.get("/").data.decode("utf-8")
 
 
+def test_cache_memoize_timeout_dynamic_via_callable_timeout(app, cache):
+    @app.route("/")
+    @cache.memoize(
+        # This should override the timeout to be 2 seconds
+        timeout=lambda rv: 2
+        if isinstance(rv, Response)
+        else 1
+    )
+    def cached_view():
+        return make_response(str(time.time()))
+
+    tc = app.test_client()
+    rv1 = tc.get("/")
+    time1 = rv1.data.decode("utf-8")
+
+    time.sleep(1)  # after 1 second, cache is still active
+    assert time1 == tc.get("/").data.decode("utf-8")
+
+    time.sleep(1)  # after 2 seconds, cache is not still active
+    assert time1 != tc.get("/").data.decode("utf-8")
+
+
+def test_cache_cached_reponse_overrides_callable_timeout(app, cache):
+    @app.route("/")
+    @cache.cached(timeout=lambda rv: 1)  # timeout to be be overridden by CachedResponse
+    def cached_view():
+        # This should override the timeout to be 2 seconds
+        return CachedResponse(response=make_response(str(time.time())), timeout=2)
+
+    tc = app.test_client()
+    rv1 = tc.get("/")
+    time1 = rv1.data.decode("utf-8")
+
+    time.sleep(1)  # after 1 second, cache is still active
+    assert time1 == tc.get("/").data.decode("utf-8")
+
+    time.sleep(1)  # after 2 seconds, cache is not still active
+    assert time1 != tc.get("/").data.decode("utf-8")
+
+
 def test_generate_cache_key_from_query_string(app, cache):
     """Test the _make_cache_key_query_string() cache key maker.