diff --git a/docs/_newsfragments/2629.misc.rst b/docs/_newsfragments/2629.misc.rst new file mode 100644 index 000000000..d8ffdc809 --- /dev/null +++ b/docs/_newsfragments/2629.misc.rst @@ -0,0 +1 @@ +Improved typing for Falcon's internal LRU cache decorators. diff --git a/falcon/media/handlers.py b/falcon/media/handlers.py index 85081863e..e148c17e2 100644 --- a/falcon/media/handlers.py +++ b/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 diff --git a/falcon/testing/client.py b/falcon/testing/client.py index 55212d814..5465a01ac 100644 --- a/falcon/testing/client.py +++ b/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), diff --git a/falcon/util/misc.py b/falcon/util/misc.py index d6d301a6e..2c2199741 100644 --- a/falcon/util/misc.py +++ b/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: