Use ParamSpec for wrapped signatures

.mypy.ini

@@ -0,0 +1,24 @@
+[mypy]
+files = async_lru, tests
+check_untyped_defs = True
+follow_imports_for_stubs = True
+disallow_any_decorated = True
+disallow_any_generics = True
+disallow_any_unimported = True
+disallow_incomplete_defs = True
+disallow_subclassing_any = True
+disallow_untyped_calls = True
+disallow_untyped_decorators = True
+disallow_untyped_defs = True
+enable_error_code = ignore-without-code, possibly-undefined, redundant-expr, redundant-self, truthy-bool, truthy-iterable, unused-awaitable
+implicit_reexport = False
+no_implicit_optional = True
+pretty = True
+show_column_numbers = True
+show_error_codes = True
+strict_equality = True
+warn_incomplete_stub = True
+warn_redundant_casts = True
+warn_return_any = True
+warn_unreachable = True
+warn_unused_ignores = True

async_lru/__init__.py

@@ -22,6 +22,12 @@
 )
 
 
+if sys.version_info >= (3, 10):
+    from typing import ParamSpec
+else:
+    from typing_extensions import ParamSpec
+
+
 if sys.version_info >= (3, 11):
     from typing import Self
 else:
@@ -35,9 +41,10 @@
 
 _T = TypeVar("_T")
 _R = TypeVar("_R")
+_P = ParamSpec("_P")
 _Coro = Coroutine[Any, Any, _R]
-_CB = Callable[..., _Coro[_R]]
-_CBP = Union[_CB[_R], "partial[_Coro[_R]]", "partialmethod[_Coro[_R]]"]
+_CB = Callable[_P, _Coro[_R]]
+_CBP = Union[_CB[_P, _R], "partial[_Coro[_R]]", "partialmethod[_Coro[_R]]"]
 
 
 @final
@@ -61,10 +68,10 @@ def cancel(self) -> None:
 
 
 @final
-class _LRUCacheWrapper(Generic[_R]):
+class _LRUCacheWrapper(Generic[_P, _R]):
     def __init__(
         self,
-        fn: _CB[_R],
+        fn: _CB[_P, _R],
         maxsize: Optional[int],
         typed: bool,
         ttl: Optional[float],
@@ -188,7 +195,7 @@ def _task_done_callback(
 
         fut.set_result(task.result())
 
-    async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
+    async def __call__(self, /, *fn_args: _P.args, **fn_kwargs: _P.kwargs) -> _R:
         if self.__closed:
             raise RuntimeError(f"alru_cache is closed for {self}")
 
@@ -207,7 +214,7 @@ async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
 
         fut = loop.create_future()
         coro = self.__wrapped__(*fn_args, **fn_kwargs)
-        task: asyncio.Task[_R] = loop.create_task(coro)
+        task = loop.create_task(coro)
         self.__tasks.add(task)
         task.add_done_callback(partial(self._task_done_callback, fut, key))
 
@@ -222,18 +229,18 @@ async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
 
     def __get__(
         self, instance: _T, owner: Optional[Type[_T]]
-    ) -> Union[Self, "_LRUCacheWrapperInstanceMethod[_R, _T]"]:
+    ) -> Union[Self, "_LRUCacheWrapperInstanceMethod[_P, _R, _T]"]:
         if owner is None:
             return self
         else:
             return _LRUCacheWrapperInstanceMethod(self, instance)
 
 
 @final
-class _LRUCacheWrapperInstanceMethod(Generic[_R, _T]):
+class _LRUCacheWrapperInstanceMethod(Generic[_P, _R, _T]):
     def __init__(
         self,
-        wrapper: _LRUCacheWrapper[_R],
+        wrapper: _LRUCacheWrapper[_P, _R],
         instance: _T,
     ) -> None:
         try:
@@ -284,16 +291,16 @@ def cache_info(self) -> _CacheInfo:
     def cache_parameters(self) -> _CacheParameters:
         return self.__wrapper.cache_parameters()
 
-    async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
-        return await self.__wrapper(self.__instance, *fn_args, **fn_kwargs)
+    async def __call__(self, /, *fn_args: _P.args, **fn_kwargs: _P.kwargs) -> _R:
+        return await self.__wrapper(self.__instance, *fn_args, **fn_kwargs)  # type: ignore[arg-type]
 
 
 def _make_wrapper(
     maxsize: Optional[int],
     typed: bool,
     ttl: Optional[float] = None,
-) -> Callable[[_CBP[_R]], _LRUCacheWrapper[_R]]:
-    def wrapper(fn: _CBP[_R]) -> _LRUCacheWrapper[_R]:
+) -> Callable[[_CBP[_P, _R]], _LRUCacheWrapper[_P, _R]]:
+    def wrapper(fn: _CBP[_P, _R]) -> _LRUCacheWrapper[_P, _R]:
         origin = fn
 
         while isinstance(origin, (partial, partialmethod)):
@@ -306,7 +313,7 @@ def wrapper(fn: _CBP[_R]) -> _LRUCacheWrapper[_R]:
         if hasattr(fn, "_make_unbound_method"):
             fn = fn._make_unbound_method()
 
-        return _LRUCacheWrapper(cast(_CB[_R], fn), maxsize, typed, ttl)
+        return _LRUCacheWrapper(cast(_CB[_P, _R], fn), maxsize, typed, ttl)
 
     return wrapper
 
@@ -317,28 +324,30 @@ def alru_cache(
     typed: bool = False,
     *,
     ttl: Optional[float] = None,
-) -> Callable[[_CBP[_R]], _LRUCacheWrapper[_R]]:
+) -> Callable[[_CBP[_P, _R]], _LRUCacheWrapper[_P, _R]]:
     ...
 
 
 @overload
 def alru_cache(
-    maxsize: _CBP[_R],
+    maxsize: _CBP[_P, _R],
     /,
-) -> _LRUCacheWrapper[_R]:
+) -> _LRUCacheWrapper[_P, _R]:
     ...
 
 
 def alru_cache(
-    maxsize: Union[Optional[int], _CBP[_R]] = 128,
+    maxsize: Union[Optional[int], _CBP[_P, _R]] = 128,
     typed: bool = False,
     *,
     ttl: Optional[float] = None,
-) -> Union[Callable[[_CBP[_R]], _LRUCacheWrapper[_R]], _LRUCacheWrapper[_R]]:
+) -> Union[
+    Callable[[_CBP[_P, _R]], _LRUCacheWrapper[_P, _R]], _LRUCacheWrapper[_P, _R]
+]:
     if maxsize is None or isinstance(maxsize, int):
         return _make_wrapper(maxsize, typed, ttl)
     else:
-        fn = cast(_CB[_R], maxsize)
+        fn = maxsize
 
         if callable(fn) or hasattr(fn, "_make_unbound_method"):
             return _make_wrapper(128, False, None)(fn)

setup.cfg

@@ -79,8 +79,3 @@ junit_family=xunit2
 asyncio_mode=auto
 timeout=15
 xfail_strict = true
-
-[mypy]
-strict=True
-pretty=True
-packages=async_lru, tests

tests/conftest.py

@@ -1,15 +1,26 @@
+import sys
 from functools import _CacheInfo
-from typing import Callable
+from typing import Callable, TypeVar
 
 import pytest
 
-from async_lru import _R, _LRUCacheWrapper
+from async_lru import _LRUCacheWrapper
+
+
+if sys.version_info >= (3, 10):
+    from typing import ParamSpec
+else:
+    from typing_extensions import ParamSpec
+
+
+_T = TypeVar("_T")
+_P = ParamSpec("_P")
 
 
 @pytest.fixture
-def check_lru() -> Callable[..., None]:
+def check_lru() -> Callable[..., None]:  # type: ignore[misc]
     def _check_lru(
-        wrapped: _LRUCacheWrapper[_R],
+        wrapped: _LRUCacheWrapper[_P, _T],
         *,
         hits: int,
         misses: int,