feat(typing): use ParamSpec to type LRU cache decorators (#2629)

falcon/_typing.py

@@ -35,6 +35,20 @@
     Union,
 )
 
+if sys.version_info >= (3, 10):
+    from typing import Concatenate as Concatenate
+    from typing import ParamSpec as ParamSpec
+
+    _P = ParamSpec('_P')
+else:
+    try:
+        from typing_extensions import Concatenate as Concatenate
+        from typing_extensions import ParamSpec as ParamSpec
+
+        _P = ParamSpec('_P')
+    except ImportError:  # pragma: nocover
+        _P = TypeVar('_P')  # type: ignore[assignment]
+
 # NOTE(vytas): Mypy still struggles to handle a conditional import in the EAFP
 #   fashion, so we branch on Py version instead (which it does understand).
 if sys.version_info >= (3, 11):
@@ -60,9 +74,16 @@ class _Unset(Enum):
 
 
 _T = TypeVar('_T')
+_R_co = TypeVar('_R_co', covariant=True)
 _UNSET = _Unset.UNSET
 UnsetOr = Union[Literal[_Unset.UNSET], _T]
 
+
+class _LruCacheWrapper(Protocol[_P, _R_co]):
+    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...
+    def cache_clear(self) -> None: ...
+
+
 # NOTE(vytas,jap): TypeVar's "default" argument is only available on 3.13+.
 if sys.version_info >= (3, 13):
     _ExcT = TypeVar('_ExcT', bound=Exception, default=Exception)

falcon/http_error.py

@@ -161,8 +161,7 @@ def status_code(self) -> int:
         """HTTP status code normalized from the ``status`` argument passed
         to the initializer.
         """  # noqa: D205
-        # TODO(0xMattB): Modify decorator to return proper type (see gh #2629).
-        return misc.http_status_to_code(self.status)  # type: ignore[no-any-return]
+        return misc.http_status_to_code(self.status)
 
     def to_dict(
         self, obj_type: type[MutableMapping[str, str | int | None | Link]] = dict

falcon/http_status.py

@@ -67,5 +67,4 @@ def __init__(
     @property
     def status_code(self) -> int:
         """HTTP status code normalized from :attr:`status`."""
-        # TODO(0xMattB): Modify decorator to return proper type (see PR #2629).
-        return http_status_to_code(self.status)  # type: ignore[no-any-return]
+        return http_status_to_code(self.status)

falcon/media/handlers.py

@@ -73,6 +73,8 @@ def __call__(
         self, media_type: str | None, default: str, raise_not_found: bool = True
     ) -> tuple[None, None, None] | _ResolverMethodReturnTuple: ...
 
+    def cache_clear(self) -> None: ...
+
 
 class Handlers(UserDict[str, BaseHandler]):
     """A :class:`dict`-like object that manages Internet media type handlers."""
@@ -98,14 +100,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/response.py

@@ -199,8 +199,7 @@ def status_code(self) -> int:
             if resp.status_code >= 400:
                 log.warning(f'returning error response: {resp.status_code}')
         """
-        # TODO(0xMattB): Modify decorator to return proper type (see gh #2629).
-        return http_status_to_code(self.status)  # type: ignore[no-any-return]
+        return http_status_to_code(self.status)
 
     @status_code.setter
     def status_code(self, value: int) -> None:

falcon/testing/client.py

@@ -962,6 +962,12 @@ async def _simulate_request_asgi(
 
         req_event_emitter.disconnect()
         await task_req
+
+        if resp_event_collector.status is None:
+            # NOTE(AlexChen): The app is expected to emit `http.response.start`
+            #   prior to completing the request.
+            raise RuntimeError('The app did not return a response status.')
+
         return Result(
             resp_event_collector.body_chunks,
             code_to_http_status(resp_event_collector.status),

falcon/util/misc.py

@@ -33,10 +33,13 @@
 import os
 import os.path
 import re
-from typing import Any, Callable
+from typing import Any, Callable, cast, overload, TYPE_CHECKING
 import unicodedata
 
 from falcon import status_codes
+from falcon._typing import _LruCacheWrapper
+from falcon._typing import _P
+from falcon._typing import _R_co
 from falcon.constants import PYPY
 from falcon.uri import encode_value
 
@@ -101,22 +104,50 @@
 # NOTE(kgriffs,vytas): This is tested in the PyPy gate but we do not want devs
 #   to have to install PyPy to check coverage on their workstations, so we use
 #   the nocover pragma here.
+@overload
 def _lru_cache_nop(
-    maxsize: int,
-) -> Callable[[Callable[..., Any]], Callable[..., Any]]:  # pragma: nocover
-    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+    maxsize: Callable[_P, _R_co],
+) -> _LruCacheWrapper[_P, _R_co]: ...  # pragma: nocover
+
+
+@overload
+def _lru_cache_nop(
+    maxsize: int | None = 128, typed: bool = False
+) -> Callable[
+    [Callable[_P, _R_co]], _LruCacheWrapper[_P, _R_co]
+]: ...  # pragma: nocover
+
+
+def _lru_cache_nop(maxsize: Any = 128, typed: bool = False) -> Any:  # pragma: nocover
+    def decorator(func: Callable[_P, _R_co]) -> _LruCacheWrapper[_P, _R_co]:
         # 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
+        func.cache_clear = lambda: None  # type: ignore[attr-defined]
 
-        return func
+        return cast(_LruCacheWrapper[_P, _R_co], func)
+
+    if callable(maxsize):
+        return decorator(maxsize)
 
     return decorator
 
 
 # PERF(kgriffs): Using lru_cache is slower on PyPy when the wrapped
 #   function is just doing a few non-IO operations.
-if PYPY:
+if TYPE_CHECKING:
+
+    @overload
+    def _lru_cache_for_simple_logic(
+        maxsize: Callable[_P, _R_co],
+    ) -> _LruCacheWrapper[_P, _R_co]: ...
+
+    @overload
+    def _lru_cache_for_simple_logic(
+        maxsize: int | None = 128, typed: bool = False
+    ) -> Callable[[Callable[_P, _R_co]], _LruCacheWrapper[_P, _R_co]]: ...
+
+    def _lru_cache_for_simple_logic(maxsize: Any = 128, typed: bool = False) -> Any: ...
+elif PYPY:
     _lru_cache_for_simple_logic = _lru_cache_nop  # pragma: nocover
 else:
     _lru_cache_for_simple_logic = functools.lru_cache

tests/asgi/test_testing_asgi.py

@@ -155,6 +155,28 @@ def test_immediate_disconnect():
         client.simulate_get('/', asgi_disconnect_ttl=0)
 
 
+async def test_app_returns_no_response_status():
+    from falcon.asgi_spec import ScopeType
+
+    async def silent_app(scope, receive, send):
+        if scope['type'] == ScopeType.LIFESPAN:
+            while True:
+                event = await receive()
+                if event['type'] == 'lifespan.startup':
+                    await send({'type': 'lifespan.startup.complete'})
+                elif event['type'] == 'lifespan.shutdown':
+                    await send({'type': 'lifespan.shutdown.complete'})
+                    return
+
+        return
+
+    conductor = testing.ASGIConductor(silent_app)
+
+    async with conductor:
+        with pytest.raises(RuntimeError):
+            await conductor.simulate_get('/')
+
+
 @pytest.mark.parametrize(
     'path, expected',
     [

tests/test_typing.py

@@ -6,7 +6,9 @@
 
 import falcon
 import falcon.asgi
+import falcon.media
 import falcon.testing
+import falcon.util.misc
 
 
 @dataclass
@@ -220,3 +222,32 @@ def _exercise_app(app: falcon.App[Any, Any]) -> None:
         'fancy': True,
         'title': '404 Not Found',
     }
+
+
+def test_lru_cache_typing() -> None:
+    @falcon.util.misc._lru_cache_for_simple_logic(maxsize=16)
+    def add(a: int, b: int) -> int:
+        return a + b
+
+    @falcon.util.misc._lru_cache_nop(maxsize=16)
+    def concat(a: str, b: str) -> str:
+        return a + b
+
+    res_add: int = add(1, 2)
+    assert res_add == 3
+    add.cache_clear()
+
+    res_concat: str = concat('x', 'y')
+    assert res_concat == 'xy'
+    concat.cache_clear()
+
+    code: int = falcon.http_status_to_code(falcon.HTTP_200)
+    assert code == 200
+    falcon.http_status_to_code.cache_clear()
+
+    status: str = falcon.code_to_http_status(200)
+    assert status == falcon.HTTP_200
+    falcon.code_to_http_status.cache_clear()
+
+    handlers = falcon.media.Handlers()
+    handlers._resolve.cache_clear()