Use ParamSpec for wrapped signatures

.github/workflows/ci-cd.yml

@@ -7,9 +7,6 @@ on:
  - '[0-9].[0-9]+' # matches to backport branches, e.g. 3.6
  tags: [ 'v*' ]
  pull_request:
- branches:
- - master
- - '[0-9].[0-9]+'
 
 
 jobs:

.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

@@ -4,7 +4,6 @@
 import sys
 from functools import _CacheInfo, _make_key, partial, partialmethod
 from typing import (
- Any,
  Callable,
  Coroutine,
  Generic,
@@ -16,12 +15,17 @@
  TypedDict,
  TypeVar,
  Union,
- cast,
  final,
  overload,
 )
 
 
+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:
@@ -38,9 +42,7 @@
 
 _T = TypeVar("_T")
 _R = TypeVar("_R")
-_Coro = Coroutine[Any, Any, _R]
-_CB = Callable[..., _Coro[_R]]
-_CBP = Union[_CB[_R], "partial[_Coro[_R]]", "partialmethod[_Coro[_R]]"]
+_P = ParamSpec("_P")
 
 
 @final
@@ -64,10 +66,10 @@ def cancel(self) -> None:
 
 
 @final
-class _LRUCacheWrapper(Generic[_R]):
+class _LRUCacheWrapper(Generic[_P, _R]):
  def __init__(
  self,
- fn: _CB[_R],
+ fn: Callable[_P, Coroutine[object, object, _R]],
  maxsize: Optional[int],
  typed: bool,
  ttl: Optional[float],
@@ -110,7 +112,7 @@ def __init__(
  self.__misses = 0
  self.__tasks: Set["asyncio.Task[_R]"] = set()
 
- def cache_invalidate(self, /, *args: Hashable, **kwargs: Any) -> bool:
+ def cache_invalidate(self, /, *args: _P.args, **kwargs: _P.kwargs) -> bool:
  key = _make_key(args, kwargs, self.__typed)
 
  cache_item = self.__cache.pop(key, None)
@@ -192,7 +194,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}")
 
@@ -211,7 +213,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))
 
@@ -224,20 +226,30 @@ async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
  self._cache_miss(key)
  return await asyncio.shield(fut)
 
+ @overload
+ def __get__(self, instance: _T, owner: None) -> Self:
+ ...
+
+ @overload
+ def __get__(
+ self, instance: _T, owner: Type[_T]
+ ) -> "_LRUCacheWrapperInstanceMethod[_P, _R, _T]":
+ ...
+
  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:
@@ -272,7 +284,7 @@ def __init__(
  self.__instance = instance
  self.__wrapper = wrapper
 
- def cache_invalidate(self, /, *args: Hashable, **kwargs: Any) -> bool:
+ def cache_invalidate(self, /, *args: _P.args, **kwargs: _P.kwargs) -> bool:
  return self.__wrapper.cache_invalidate(self.__instance, *args, **kwargs)
 
  def cache_clear(self) -> None:
@@ -289,16 +301,18 @@ 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[[Callable[_P, Coroutine[object, object, _R]]], _LRUCacheWrapper[_P, _R]]:
+ def wrapper(
+ fn: Callable[_P, Coroutine[object, object, _R]]
+ ) -> _LRUCacheWrapper[_P, _R]:
  origin = fn
 
  while isinstance(origin, (partial, partialmethod)):
@@ -311,7 +325,7 @@ def wrapper(fn: _CBP[_R]) -> _LRUCacheWrapper[_R]:
  if hasattr(fn, "_make_unbound_method"):
  fn = fn._make_unbound_method()
 
- wrapper = _LRUCacheWrapper(cast(_CB[_R], fn), maxsize, typed, ttl)
+ wrapper = _LRUCacheWrapper(fn, maxsize, typed, ttl)
  if sys.version_info >= (3, 12):
  wrapper = inspect.markcoroutinefunction(wrapper)
  return wrapper
@@ -325,30 +339,34 @@ def alru_cache(
  typed: bool = False,
  *,
  ttl: Optional[float] = None,
-) -> Callable[[_CBP[_R]], _LRUCacheWrapper[_R]]:
+) -> Callable[[Callable[_P, Coroutine[object, object, _R]]], _LRUCacheWrapper[_P, _R]]:
  ...
 
 
 @overload
 def alru_cache(
- maxsize: _CBP[_R],
+ maxsize: Callable[_P, Coroutine[object, object, _R]],
  /,
-) -> _LRUCacheWrapper[_R]:
+) -> _LRUCacheWrapper[_P, _R]:
  ...
 
 
 def alru_cache(
- maxsize: Union[Optional[int], _CBP[_R]] = 128,
+ maxsize: Union[Optional[int], Callable[_P, Coroutine[object, object, _R]]] = 128,
  typed: bool = False,
  *,
  ttl: Optional[float] = None,
-) -> Union[Callable[[_CBP[_R]], _LRUCacheWrapper[_R]], _LRUCacheWrapper[_R]]:
+) -> Union[
+ Callable[[Callable[_P, Coroutine[object, object, _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"):
+ # partialmethod is not callable() at runtime.
+ if callable(fn) or hasattr(fn, "_make_unbound_method"): # type: ignore[unreachable]
  return _make_wrapper(128, False, None)(fn)
 
  raise NotImplementedError(f"{fn!r} decorating is not supported")

setup.cfg

@@ -82,8 +82,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(
- wrapped: _LRUCacheWrapper[_R],
+ wrapped: _LRUCacheWrapper[_P, _T],
  *,
  hits: int,
  misses: int,

tests/test_basic.py

@@ -88,6 +88,21 @@ async def coro(val: int) -> int:
  assert await coro_wrapped2() == 2
 
 
+async def test_alru_cache_partial_typing() -> None:
+ """Test that mypy produces call-arg errors correctly."""
+
+ async def coro(val: int) -> int:
+ return val
+
+ coro_wrapped1 = alru_cache(coro)
+ with pytest.raises(TypeError):
+ await coro_wrapped1(1, 1) # type: ignore[call-arg]
+
+ coro_wrapped2 = alru_cache(partial(coro, 2))
+ with pytest.raises(TypeError):
+ await coro_wrapped2(4) == 2 # type: ignore[call-arg]
+
+
 async def test_alru_cache_await_same_result_async(
  check_lru: Callable[..., None]
 ) -> None:

tests/test_exception.py

@@ -33,7 +33,7 @@ async def coro(val: int) -> None:
  reason="Memory leak is not fixed for PyPy3.9",
  condition=sys.implementation.name == "pypy",
 )
-async def test_alru_exception_reference_cleanup(check_lru: Callable[..., None]) -> None:
+async def test_alru_exception_reference_cleanup(check_lru: Callable[..., None]) -> None: # type: ignore[misc]
  class CustomClass:
  ...