Use ParamSpec for LRU cache decorators

docs/_newsfragments/2629.misc.rst

@@ -0,0 +1 @@
+Improved typing for Falcon's internal LRU cache decorators.

falcon/media/handlers.py

@@ -6,6 +6,7 @@
 import functools
 from typing import (
     Any,
+    Callable,
     cast,
     Literal,
     NoReturn,
@@ -56,6 +57,8 @@ def _raise(self, *args: Any, **kwargs: Any) -> NoReturn:
 
 
 class ResolverMethod(Protocol):
+    cache_clear: Callable[[], None]
+
     @overload
     def __call__(
         self, media_type: str | None, default: str, raise_not_found: Literal[False]
@@ -98,14 +101,14 @@ def __setitem__(self, key: str, value: BaseHandler) -> None:
         # NOTE(kgriffs): When the mapping changes, we do not want to use a
         #   cached handler from the previous mapping, in case it was
         #   replaced.
-        self._resolve.cache_clear()  # type: ignore[attr-defined]
+        self._resolve.cache_clear()
 
     def __delitem__(self, key: str) -> None:
         super().__delitem__(key)
 
         # NOTE(kgriffs): Similar to __setitem__(), we need to avoid resolving
         #   to a cached handler that was removed.
-        self._resolve.cache_clear()  # type: ignore[attr-defined]
+        self._resolve.cache_clear()
 
     def _create_resolver(self) -> ResolverMethod:
         # PERF(kgriffs): Under PyPy the LRU is relatively expensive as compared

falcon/testing/client.py

@@ -963,6 +963,12 @@ async def _simulate_request_asgi(
 
         req_event_emitter.disconnect()
         await task_req
+
+        if resp_event_collector.status is None:
+            # NOTE(kgriffs): An immediate disconnect was simulated, and so
+            #   the app could not return a status.
+            raise ConnectionError('An immediate disconnect was simulated.')
+
         return Result(
             resp_event_collector.body_chunks,
             code_to_http_status(resp_event_collector.status),

falcon/util/misc.py

@@ -33,7 +33,7 @@
 import os
 import os.path
 import re
-from typing import Any, Callable
+from typing import Any, Callable, cast, TYPE_CHECKING, TypeVar
 import unicodedata
 
 from falcon import status_codes
@@ -87,6 +87,25 @@
     datetime.datetime.now, datetime.timezone.utc
 )
 
+_T = TypeVar('_T')
+_T_co = TypeVar('_T_co', covariant=True)
+
+if TYPE_CHECKING:
+    from typing import ParamSpec, Protocol
+
+    _P = ParamSpec('_P')
+
+    class _CachedCallable(Protocol[_P, _T_co]):
+        cache_clear: Callable[[], None]
+
+        def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _T_co: ...
+
+    class _LruCacheDecorator(Protocol):
+        def __call__(
+            self, maxsize: int
+        ) -> Callable[[Callable[_P, _T]], _CachedCallable[_P, _T]]: ...
+
+
 # The above aliases were not underscored prior to Falcon 3.1.2.
 strptime: Callable[[str, str], datetime.datetime] = deprecated(
     'This was a private alias local to this module; '
@@ -103,23 +122,25 @@
 #   the nocover pragma here.
 def _lru_cache_nop(
     maxsize: int,
-) -> Callable[[Callable[..., Any]], Callable[..., Any]]:  # pragma: nocover
-    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+) -> Callable[[Callable[_P, _T]], _CachedCallable[_P, _T]]:  # pragma: nocover
+    def decorator(func: Callable[_P, _T]) -> _CachedCallable[_P, _T]:
         # NOTE(kgriffs): Partially emulate the lru_cache protocol; only add
         #   cache_info() later if/when it becomes necessary.
-        func.cache_clear = lambda: None  # type: ignore
+        cached_func = cast('_CachedCallable[_P, _T]', func)
+        cached_func.cache_clear = lambda: None
 
-        return func
+        return cached_func
 
     return decorator
 
 
 # PERF(kgriffs): Using lru_cache is slower on PyPy when the wrapped
 #   function is just doing a few non-IO operations.
+_lru_cache_for_simple_logic: '_LruCacheDecorator'
 if PYPY:
     _lru_cache_for_simple_logic = _lru_cache_nop  # pragma: nocover
 else:
-    _lru_cache_for_simple_logic = functools.lru_cache
+    _lru_cache_for_simple_logic = cast('_LruCacheDecorator', functools.lru_cache)
 
 
 def is_python_func(func: Callable[..., Any] | Any) -> bool: