100% spec compliance, works with mypy and pyright

Author: mikeshardmind

_misc/_type_assertions.py

@@ -26,21 +26,23 @@
 
 
 @lrutaskcache(ttl=60, maxsize=512)
-async def t1(a: int, b: str, *, x: bytes) -> str: ...
+async def t1(a: int, b: str, *, x: bytes) -> str:
+    return ""
 
 
 assert_type(t1(1, "a", x=b""), asyncio.Task[str])
 
 
 @lrucorocache(ttl=60, maxsize=512)
-async def t2(a: int, b: str, *, x: bytes) -> str: ...
+async def t2(a: int, b: str, *, x: bytes) -> str:
+    return ""
 
 
-_ = assert_type(t2(1, "a", x=b""), Coroutine[Any, Any, str])
+_t2_type: Coroutine[Any, Any, str] = assert_type(t2(1, "a", x=b""), Coroutine[Any, Any, str])
 
 
 def gen() -> Generator[int]:
     yield from range(10)
 
 
-_ = assert_type(sync_to_async_gen_noctx(gen), AsyncGenerator[int])
+_gen_assert_type: AsyncGenerator[int] = assert_type(sync_to_async_gen_noctx(gen), AsyncGenerator[int])

readme.md

@@ -93,9 +93,6 @@ The project currently uses pyright for development, and both pyright and mypy
 when ensuring the public api surface is well-typed and compatible with strict
 use of typechecking. The configurations used are in pyproject.toml.
 
-Mypy is known to be broken currently, for the cache decorators
-[discards type information in Callable aliases](https://github.com/python/mypy/issues/18842)
-
 Use of Any in a few places is *intentional* for internals.
 
 ### 5. Threading and multiple event loops

src/async_utils/__init__.py

@@ -9,7 +9,7 @@
 __author__ = "Michael Hall"
 __license__ = "Apache-2.0"
 __copyright__ = "Copyright 2020-Present Michael Hall"
-__version__ = "2025.04.9b"
+__version__ = "2025.04.24b"
 
 import os
 import sys

src/async_utils/corofunc_cache.py

@@ -36,7 +36,18 @@
 #: Warning: Mutations will impact callsite, return new objects as needed.
 type CacheTransformer = Callable[[tuple[t.Any, ...], dict[str, t.Any]], _CT_RET]
 
-type Deco[**P, R] = Callable[[CoroLike[P, R]], CoroFunc[P, R]]
+# This is the only way to return a generic function not dependent on
+# type-checker specific behavior where the generic is deferred until application
+# of the function.
+TYPE_CHECKING = False
+if TYPE_CHECKING:
+    from typing import Protocol
+    class Deco(Protocol):
+        def __call__[**P, R](self, c: CoroLike[P, R], /) -> CoroFunc[P, R]:  ...
+else:
+    # This branch is here for something reasonable to exist at runtime
+    # Not importing typing at runtime
+    type Deco[**P, R] = Callable[[CoroLike[P, R]], CoroFunc[P, R]]
 
 
 def _chain_fut[R](c_fut: cf.Future[R], a_fut: asyncio.Future[R]) -> None:
@@ -48,11 +59,11 @@ def _chain_fut[R](c_fut: cf.Future[R], a_fut: asyncio.Future[R]) -> None:
         c_fut.set_result(a_fut.result())
 
 
-def corocache[**P, R](
+def corocache(
     ttl: float | None = None,
     *,
     cache_transform: CacheTransformer | None = None,
-) -> Deco[P, R]:
+) -> Deco:
     """Cache the results of the decorated coroutine.
 
     This is less powerful than the version in task_cache.py but may work better
@@ -85,7 +96,7 @@ def corocache[**P, R](
         def key_func(args: tuple[t.Any, ...], kwds: dict[t.Any, t.Any], /) -> Hashable:
             return make_key(*cache_transform(args, kwds))
 
-    def wrapper(coro: CoroLike[P, R]) -> CoroFunc[P, R]:
+    def wrapper[**P, R](coro: CoroLike[P, R], /) -> CoroFunc[P, R]:
         internal_cache: dict[Hashable, cf.Future[R]] = {}
         internal_taskset: set[asyncio.Task[R]] = set()
 
@@ -120,12 +131,12 @@ async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
     return wrapper
 
 
-def lrucorocache[**P, R](
+def lrucorocache(
     ttl: float | None = None,
     maxsize: int = 1024,
     *,
     cache_transform: CacheTransformer | None = None,
-) -> Deco[P, R]:
+) -> Deco:
     """Cache the results of the decorated coroutine.
 
     This is less powerful than the version in task_cache.py but may work better
@@ -164,7 +175,7 @@ def lrucorocache[**P, R](
         def key_func(args: tuple[t.Any, ...], kwds: dict[t.Any, t.Any], /) -> Hashable:
             return make_key(*cache_transform(args, kwds))
 
-    def wrapper(coro: CoroLike[P, R]) -> CoroFunc[P, R]:
+    def wrapper[**P, R](coro: CoroLike[P, R], /) -> CoroFunc[P, R]:
         internal_cache: LRU[Hashable, cf.Future[R]] = LRU(maxsize)
         internal_taskset: set[asyncio.Task[R]] = set()
 

src/async_utils/task_cache.py

@@ -38,7 +38,18 @@
 #: Warning: Mutations will impact callsite, return new objects as needed.
 type CacheTransformer = Callable[[tuple[t.Any, ...], dict[str, t.Any]], _CT_RET]
 
-type Deco[**P, R] = Callable[[TaskCoroFunc[P, R]], TaskFunc[P, R]]
+# This is the only way to return a generic function not dependent on
+# type-checker specific behavior where the generic is deferred until application
+# of the function.
+TYPE_CHECKING = False
+if TYPE_CHECKING:
+    from typing import Protocol
+    class Deco(Protocol):
+        def __call__[**P, R](self, c: TaskCoroFunc[P, R], /) -> TaskFunc[P, R]:  ...
+else:
+    # This branch is here for something reasonable to exist at runtime
+    # Not importing typing at runtime
+    type Deco[**P, R] = Callable[[TaskCoroFunc[P, R]], TaskFunc[P, R]]
 
 # Non-annotation assignments for transformed functions
 _WRAP_ASSIGN = ("__module__", "__name__", "__qualname__", "__doc__")
@@ -58,7 +69,7 @@ class _WrappedSignature[**P, R]:
     # as func.__signature__
     # Known working: py 3.12.0 - py3.14a6 range inclusive
     def __init__(self, f: TaskCoroFunc[P, R], w: TaskFunc[P, R]) -> None:
-        self._f = f
+        self._f: Callable[..., t.Any] = f  # anotation needed for inspect use below....
         self._w = w
         self._sig: t.Any | None = None
 
@@ -102,11 +113,11 @@ async def _await[R](fut: asyncio.Future[R]) -> R:
     return await fut
 
 
-def taskcache[**P, R](
+def taskcache(
     ttl: float | None = None,
     *,
     cache_transform: CacheTransformer | None = None,
-) -> Deco[P, R]:
+) -> Deco:
     """Cache the results of the decorated coroutine.
 
     Decorator to modify coroutine functions to instead act as functions
@@ -145,7 +156,7 @@ def taskcache[**P, R](
         def key_func(args: tuple[t.Any, ...], kwds: dict[t.Any, t.Any], /) -> Hashable:
             return make_key(*cache_transform(args, kwds))
 
-    def wrapper(coro: TaskCoroFunc[P, R]) -> TaskFunc[P, R]:
+    def wrapper[**P, R](coro: TaskCoroFunc[P, R], /) -> TaskFunc[P, R]:
         internal_cache: dict[Hashable, cf.Future[R]] = {}
 
         def _internal_cache_evict(key: Hashable, _ignored_task: object) -> None:
@@ -180,12 +191,12 @@ def wrapped(*args: P.args, **kwargs: P.kwargs) -> asyncio.Task[R]:
     return wrapper
 
 
-def lrutaskcache[**P, R](
+def lrutaskcache(
     ttl: float | None = None,
     maxsize: int = 1024,
     *,
     cache_transform: CacheTransformer | None = None,
-) -> Deco[P, R]:
+) -> Deco:
     """Cache the results of the decorated coroutine.
 
     Decorator to modify coroutine functions to instead act as functions
@@ -231,7 +242,7 @@ def lrutaskcache[**P, R](
         def key_func(args: tuple[t.Any, ...], kwds: dict[t.Any, t.Any], /) -> Hashable:
             return make_key(*cache_transform(args, kwds))
 
-    def wrapper(coro: TaskCoroFunc[P, R]) -> TaskFunc[P, R]:
+    def wrapper[**P, R](coro: TaskCoroFunc[P, R], /) -> TaskFunc[P, R]:
         internal_cache: LRU[Hashable, cf.Future[R]] = LRU(maxsize)
 
         def _internal_cache_evict(key: Hashable, _ignored_task: object) -> None: