Add feature: Call with freshness threshold (#262)

.gitignore

@@ -106,3 +106,5 @@ Pipfile.lock
 .DS_Store
 
 tags
+
+.cursor

README.rst

@@ -51,6 +51,7 @@ Features
 * Local caching using pickle files.
 * Cross-machine caching using MongoDB.
 * Thread-safety.
+* **Per-call max age:** Specify a maximum age for cached values per call.
 
 Cachier is **NOT**:
 
@@ -233,6 +234,27 @@ Per-function call arguments
 
 Cachier also accepts several keyword arguments in the calls of the function it wraps rather than in the decorator call, allowing you to modify its behaviour for a specific function call.
 
+**Max Age (max_age)**
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+You can specify a maximum allowed age for a cached value on a per-call basis using the `max_age` keyword argument. If the cached value is older than this threshold, a recalculation is triggered. This is in addition to the `stale_after` parameter set at the decorator level; the strictest (smallest) threshold is enforced.
+
+.. code-block:: python
+
+  from datetime import timedelta
+  from cachier import cachier
+
+  @cachier(stale_after=timedelta(days=3))
+  def add(a, b):
+      return a + b
+
+  # Use a per-call max age:
+  result = add(1, 2, max_age=timedelta(seconds=10))  # Only use cache if value is <10s old
+
+**How it works:**
+- The effective max age threshold is the minimum of `stale_after` (from the decorator) and `max_age` (from the call).
+- If the cached value is older than this threshold, a new calculation is triggered and the cache is updated.
+- If not, the cached value is returned as usual.
+
 Ignore Cache
 ~~~~~~~~~~~~
 

pyproject.toml

@@ -172,6 +172,7 @@ markers = [
   "memory: test the memory core",
   "pickle: test the pickle core",
   "sql: test the SQL core",
+  "maxage: test the max_age functionality",
 ]
 
 # --- coverage ---

src/cachier/core.py

@@ -31,6 +31,7 @@
 
 MAX_WORKERS_ENVAR_NAME = "CACHIER_MAX_WORKERS"
 DEFAULT_MAX_WORKERS = 8
+ZERO_TIMEDELTA = timedelta(seconds=0)
 
 
 def _max_workers():
@@ -225,8 +226,31 @@ def cachier(
     def _cachier_decorator(func):
         core.set_func(func)
 
-        @wraps(func)
-        def func_wrapper(*args, **kwds):
+        # ---
+        # MAINTAINER NOTE: max_age parameter
+        #
+        # The _call function below supports a per-call 'max_age' parameter,
+        # allowing users to specify a maximum allowed age for a cached value.
+        # If the cached value is older than 'max_age',
+        # a recalculation is triggered. This is in addition to the
+        # per-decorator 'stale_after' parameter.
+        #
+        # The effective staleness threshold is the minimum of 'stale_after'
+        # and 'max_age' (if provided).
+        # This ensures that the strictest max age requirement is enforced.
+        #
+        # The main function wrapper is a standard function that passes
+        # *args and **kwargs to _call. By default, max_age is None,
+        # so only 'stale_after' is considered unless overridden.
+        #
+        # The user-facing API exposes:
+        #   - Per-call: myfunc(..., max_age=timedelta(...))
+        #
+        # This design allows both one-off (per-call) and default
+        # (per-decorator) max age constraints.
+        # ---
+
+        def _call(*args, max_age: Optional[timedelta] = None, **kwds):
             nonlocal allow_none
             _allow_none = _update_with_defaults(allow_none, "allow_none", kwds)
             # print('Inside general wrapper for {}.'.format(func.__name__))
@@ -271,7 +295,23 @@ def func_wrapper(*args, **kwds):
             if _allow_none or entry.value is not None:
                 _print("Cached result found.")
                 now = datetime.now()
-                if now - entry.time <= _stale_after:
+                max_allowed_age = _stale_after
+                nonneg_max_age = True
+                if max_age is not None:
+                    if max_age < ZERO_TIMEDELTA:
+                        _print(
+                            "max_age is negative. "
+                            "Cached result considered stale."
+                        )
+                        nonneg_max_age = False
+                    else:
+                        max_allowed_age = (
+                            min(_stale_after, max_age)
+                            if max_age is not None
+                            else _stale_after
+                        )
+                # note: if max_age < 0, we always consider a value stale
+                if nonneg_max_age and (now - entry.time <= max_allowed_age):
                     _print("And it is fresh!")
                     return entry.value
                 _print("But it is stale... :(")
@@ -305,6 +345,14 @@ def func_wrapper(*args, **kwds):
             _print("No entry found. No current calc. Calling like a boss.")
             return _calc_entry(core, key, func, args, kwds)
 
+        # MAINTAINER NOTE: The main function wrapper is now a standard function
+        # that passes *args and **kwargs to _call. This ensures that user
+        # arguments are not shifted, and max_age is only settable via keyword
+        # argument.
+        @wraps(func)
+        def func_wrapper(*args, **kwargs):
+            return _call(*args, **kwargs)
+
         def _clear_cache():
             """Clear the cache."""
             core.clear_cache()

tests/test_call_with_max_age.py

@@ -0,0 +1,134 @@
+import time
+from datetime import timedelta
+
+import pytest
+
+import cachier
+
+
+@pytest.mark.maxage
+def test_call_with_max_age():
+    @cachier.cachier()
+    def test_func(a, b):
+        return a + b
+
+    # First call: should compute and cache
+    val1 = test_func(1, 2)
+    assert val1 == 3
+    # Second call: should use cache
+    val2 = test_func(1, 2)
+    assert val2 == 3
+    # Wait for cache to become stale
+    time.sleep(1.0)
+    # Should trigger recalculation (stale)
+    val3 = test_func(1, 2, max_age=timedelta(seconds=0.5))
+    assert val3 == 3
+
+
+@pytest.mark.maxage
+def test_max_age_stricter_than_stale_after():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=2))
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    v2 = f(1)
+    assert v1 == v2  # cache hit
+    time.sleep(1)
+    v3 = f(1, max_age=timedelta(seconds=0.5))
+    assert v3 != v1  # max_age stricter, triggers recalc
+
+
+@pytest.mark.maxage
+def test_max_age_looser_than_stale_after():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=1))
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    v2 = f(1)
+    assert v1 == v2
+    time.sleep(1.1)
+    v3 = f(1, max_age=timedelta(seconds=5))
+    assert v3 != v1  # max_age looser, but stale_after still applies (stricter)
+
+
+@pytest.mark.maxage
+def test_max_age_none_defaults_to_stale_after():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=1))
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    time.sleep(1.1)
+    v2 = f(1, max_age=None)
+    assert v2 != v1  # Should trigger recalc (stale_after applies)
+
+
+@pytest.mark.maxage
+def test_negative_max_age_triggers_recalc():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=100))
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    time.sleep(0.5)  # Ensure some time has passed
+    v2 = f(1, max_age=timedelta(seconds=-1), cachier__verbose=True)
+    assert v2 != v1  # Negative max_age always triggers recalc
+
+
+@pytest.mark.maxage
+def test_max_age_zero():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=100))
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    # Add a small sleep to ensure measurable time difference on all platforms
+    time.sleep(1)
+    v2 = f(1, max_age=timedelta(seconds=0))
+    assert v2 != v1  # Zero max_age always triggers recalc
+
+
+@pytest.mark.maxage
+def test_max_age_with_next_time():
+    import time
+
+    import cachier
+
+    @cachier.cachier(stale_after=timedelta(seconds=1), next_time=True)
+    def f(x):
+        return time.time()
+
+    f.clear_cache()
+    v1 = f(1)
+    time.sleep(1.1)
+    v2 = f(1, max_age=timedelta(seconds=0.5))
+    # With next_time=True, should return stale value (v1) while
+    # triggering a recalculation in the background
+    assert v2 == v1