fix(typing): preserve decorated function type signatures with ParamSpec (#376)

* fix(typing): preserve decorated function type signatures with ParamSpec

* fix: Revert uv.lock

* fix: Preserve the KW args; New tests

* test(typing): address review comments on PR

* fix(typing): Ignore the type hints-related code from coverage

* ci: skip docformatter on pre-commit.ci

docformatter v1.7.7 transitively pulls untokenize, whose setup.py uses
ast.Constant.s (removed in Python 3.12+) and fails to install on
pre-commit.ci's runners. Skip the hook on pre-commit.ci until
docformatter v1.7.8 ships (which drops the untokenize dep). The
pre-commit GitHub Actions job still runs docformatter, so coverage of
the hook isn't lost.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: gencurrent

.pre-commit-config.yaml

@@ -6,6 +6,12 @@ ci:
   autoupdate_commit_msg: "[pre-commit.ci] pre-commit suggestions"
   autoupdate_schedule: quarterly
   # submodules: true
+  # docformatter v1.7.7 transitively pulls `untokenize`, whose setup.py
+  # uses `ast.Constant.s` (removed in Python 3.12+) and fails to install
+  # on pre-commit.ci's runners. The `pre-commit` GitHub Actions job still
+  # runs docformatter, so coverage isn't lost. Remove this once docformatter
+  # ships v1.7.8 (which drops the untokenize dep).
+  skip: [docformatter]
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks

src/cachier/core.py

@@ -16,7 +16,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta
 from functools import wraps
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, ParamSpec, Protocol, TypeVar, Union
 from warnings import warn
 
 from ._types import RedisClient, S3Client
@@ -31,6 +31,34 @@
 from .metrics import CacheMetrics, MetricsContext
 from .util import parse_bytes
 
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+_R_co = TypeVar("_R_co", covariant=True)
+
+
+class _CachierWrappedFunc(Protocol[_P, _R_co]):
+    """Callable returned by ``@cachier`` with the decorated function's signature.
+
+    Preserves the original function's parameter and return types via ``ParamSpec``
+    while also exposing the cache-management attributes attached by the decorator.
+    Per-call cachier options such as ``max_age`` and ``cachier__skip_cache`` are
+    accepted at runtime but are not surfaced in the ``__call__`` signature here;
+    PEP 612 does not permit mixing ParamSpec kwargs with additional keyword-only
+    parameters.
+
+    """
+
+    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...  # pragma: no cover
+
+    clear_cache: Callable[[], Any]
+    clear_being_calculated: Callable[[], Any]
+    aclear_cache: Callable[[], Any]
+    aclear_being_calculated: Callable[[], Any]
+    cache_dpath: Callable[[], Optional[str]]
+    precache_value: Callable[..., Any]
+    metrics: Optional[CacheMetrics]
+
+
 MAX_WORKERS_ENVAR_NAME = "CACHIER_MAX_WORKERS"
 DEFAULT_MAX_WORKERS = 8
 ZERO_TIMEDELTA = timedelta(seconds=0)
@@ -221,7 +249,7 @@ def cachier(
     allow_non_static_methods: Optional[bool] = None,
     enable_metrics: bool = False,
     metrics_sampling_rate: float = 1.0,
-):
+) -> Callable[[Callable[_P, _R]], _CachierWrappedFunc[_P, _R]]:
     """Wrap as a persistent, stale-free memoization decorator.
 
     The positional and keyword arguments to the wrapped function must be
@@ -400,7 +428,7 @@ def cachier(
     else:
         raise ValueError("specified an invalid core: %s" % backend)
 
-    def _cachier_decorator(func):
+    def _cachier_decorator(func: Callable[_P, _R]) -> _CachierWrappedFunc[_P, _R]:
         core.set_func(func)
 
         # Guard: raise TypeError when decorating an instance method unless
@@ -513,7 +541,7 @@ def _call(*args, max_age: Optional[timedelta] = None, **kwds):
             from .config import _global_params
 
             if ignore_cache or not _global_params.caching_enabled:
-                return func(args[0], **kwargs) if core.func_is_method else func(**kwargs)
+                return func(args[0], **kwargs) if core.func_is_method else func(**kwargs)  # type: ignore[call-arg]
 
             with MetricsContext(cache_metrics) as _mctx:
                 key, entry = core.get_entry((), kwargs)
@@ -629,7 +657,7 @@ async def _call_async(*args, max_age: Optional[timedelta] = None, **kwds):
             from .config import _global_params
 
             if ignore_cache or not _global_params.caching_enabled:
-                return await func(args[0], **kwargs) if core.func_is_method else await func(**kwargs)
+                return await func(args[0], **kwargs) if core.func_is_method else await func(**kwargs)  # type: ignore[call-arg,misc]
 
             with MetricsContext(cache_metrics) as _mctx:
                 key, entry = await core.aget_entry((), kwargs)
@@ -699,14 +727,14 @@ async def _call_async(*args, max_age: Optional[timedelta] = None, **kwds):
         if is_coroutine:
 
             @wraps(func)
-            async def func_wrapper(*args, **kwargs):
-                return await _call_async(*args, **kwargs)
+            async def func_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+                return await _call_async(*args, **kwargs)  # type: ignore[arg-type]
 
         else:
 
             @wraps(func)
-            def func_wrapper(*args, **kwargs):
-                return _call(*args, **kwargs)
+            def func_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
+                return _call(*args, **kwargs)  # type: ignore[arg-type]
 
         def _clear_cache():
             """Clear the cache."""
@@ -751,13 +779,13 @@ def _precache_value(*args, value_to_cache, **kwds):
             kwargs = _convert_args_kwargs(func, _is_method=core.func_is_method, args=args, kwds=kwds)
             return core.precache_value((), kwargs, value_to_cache)
 
-        func_wrapper.clear_cache = _clear_cache
-        func_wrapper.clear_being_calculated = _clear_being_calculated
-        func_wrapper.aclear_cache = _aclear_cache
-        func_wrapper.aclear_being_calculated = _aclear_being_calculated
-        func_wrapper.cache_dpath = _cache_dpath
-        func_wrapper.precache_value = _precache_value
-        func_wrapper.metrics = cache_metrics  # Expose metrics object
-        return func_wrapper
+        func_wrapper.clear_cache = _clear_cache  # type: ignore[attr-defined]
+        func_wrapper.clear_being_calculated = _clear_being_calculated  # type: ignore[attr-defined]
+        func_wrapper.aclear_cache = _aclear_cache  # type: ignore[attr-defined]
+        func_wrapper.aclear_being_calculated = _aclear_being_calculated  # type: ignore[attr-defined]
+        func_wrapper.cache_dpath = _cache_dpath  # type: ignore[attr-defined]
+        func_wrapper.precache_value = _precache_value  # type: ignore[attr-defined]
+        func_wrapper.metrics = cache_metrics  # type: ignore[attr-defined]
+        return func_wrapper  # type: ignore[return-value]
 
     return _cachier_decorator

tests/requirements.txt

@@ -7,6 +7,8 @@ pytest-rerunfailures  # for retrying flaky tests
 coverage
 pytest-cov
 birch
+# type checking
+mypy
 # to be able to run `python setup.py checkdocs`
 collective.checkdocs
 pygments