Generic ExceptionHandler type

starlette/_exception_handler.py

@@ -16,13 +16,13 @@
 )
 from starlette.websockets import WebSocket
 
-ExceptionHandlers = typing.Dict[typing.Any, ExceptionHandler]
-StatusHandlers = typing.Dict[int, ExceptionHandler]
+ExceptionHandlers = typing.Dict[typing.Any, ExceptionHandler[Exception]]
+StatusHandlers = typing.Dict[int, ExceptionHandler[Exception]]
 
 
 def _lookup_exception_handler(
     exc_handlers: ExceptionHandlers, exc: Exception
-) -> typing.Optional[ExceptionHandler]:
+) -> typing.Optional[ExceptionHandler[Exception]]:
     for cls in type(exc).__mro__:
         if cls in exc_handlers:
             return exc_handlers[cls]
@@ -69,15 +69,15 @@ async def sender(message: Message) -> None:
 
             if scope["type"] == "http":
                 nonlocal conn
-                handler = typing.cast(HTTPExceptionHandler, handler)
+                handler = typing.cast(HTTPExceptionHandler[Exception], handler)
                 conn = typing.cast(Request, conn)
                 if is_async_callable(handler):
                     response = await handler(conn, exc)
                 else:
                     response = await run_in_threadpool(handler, conn, exc)
                 await response(scope, receive, sender)
             elif scope["type"] == "websocket":
-                handler = typing.cast(WebSocketExceptionHandler, handler)
+                handler = typing.cast(WebSocketExceptionHandler[Exception], handler)
                 conn = typing.cast(WebSocket, conn)
                 if is_async_callable(handler):
                     await handler(conn, exc)

starlette/applications.py

@@ -13,7 +13,7 @@
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.routing import BaseRoute, Router
-from starlette.types import ASGIApp, ExceptionHandler, Lifespan, Receive, Scope, Send
+from starlette.types import ASGIApp, E, ExceptionHandler, Lifespan, Receive, Scope, Send
 from starlette.websockets import WebSocket
 
 AppType = typing.TypeVar("AppType", bound="Starlette")
@@ -55,7 +55,8 @@ def __init__(
         debug: bool = False,
         routes: typing.Sequence[BaseRoute] | None = None,
         middleware: typing.Sequence[Middleware] | None = None,
-        exception_handlers: typing.Mapping[typing.Any, ExceptionHandler] | None = None,
+        exception_handlers: typing.Mapping[typing.Any, ExceptionHandler[E]]
+        | None = None,
         on_startup: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
         on_shutdown: typing.Sequence[typing.Callable[[], typing.Any]] | None = None,
         lifespan: typing.Optional[Lifespan["AppType"]] = None,
@@ -139,8 +140,8 @@ def add_middleware(
 
     def add_exception_handler(
         self,
-        exc_class_or_status_code: int | typing.Type[Exception],
-        handler: ExceptionHandler,
+        exc_class_or_status_code: int | typing.Type[E],
+        handler: ExceptionHandler[E],
     ) -> None:  # pragma: no cover
         self.exception_handlers[exc_class_or_status_code] = handler
 

starlette/authentication.py

@@ -35,7 +35,7 @@ def requires(
     scopes_list = [scopes] if isinstance(scopes, str) else list(scopes)
 
     def decorator(
-        func: typing.Callable[_P, typing.Any]
+        func: typing.Callable[_P, typing.Any],
     ) -> typing.Callable[_P, typing.Any]:
         sig = inspect.signature(func)
         for idx, parameter in enumerate(sig.parameters.values()):

starlette/routing.py

@@ -55,7 +55,9 @@ def iscoroutinefunction_or_partial(obj: typing.Any) -> bool:  # pragma: no cover
 
 
 def request_response(
-    func: typing.Callable[[Request], typing.Union[typing.Awaitable[Response], Response]]
+    func: typing.Callable[
+        [Request], typing.Union[typing.Awaitable[Response], Response]
+    ],
 ) -> ASGIApp:
     """
     Takes a function or coroutine `func(request) -> response`,
@@ -78,7 +80,7 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
 
 
 def websocket_session(
-    func: typing.Callable[[WebSocket], typing.Awaitable[None]]
+    func: typing.Callable[[WebSocket], typing.Awaitable[None]],
 ) -> ASGIApp:
     """
     Takes a coroutine `func(session)`, and returns an ASGI application.

starlette/types.py

@@ -21,10 +21,10 @@
 ]
 Lifespan = typing.Union[StatelessLifespan[AppType], StatefulLifespan[AppType]]
 
+E = typing.TypeVar("E", bound=Exception, contravariant=True)
+
 HTTPExceptionHandler = typing.Callable[
-    ["Request", Exception], typing.Union["Response", typing.Awaitable["Response"]]
-]
-WebSocketExceptionHandler = typing.Callable[
-    ["WebSocket", Exception], typing.Awaitable[None]
+    ["Request", E], typing.Union["Response", typing.Awaitable["Response"]]
 ]
-ExceptionHandler = typing.Union[HTTPExceptionHandler, WebSocketExceptionHandler]
+WebSocketExceptionHandler = typing.Callable[["WebSocket", E], typing.Awaitable[None]]
+ExceptionHandler = typing.Union[HTTPExceptionHandler[E], WebSocketExceptionHandler[E]]

tests/middleware/test_base.py

@@ -499,7 +499,7 @@ async def downstream_app(scope, receive, send):
 
 
 def test_read_request_stream_in_app_after_middleware_calls_stream(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         expected = [b""]
@@ -527,7 +527,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_stream_in_app_after_middleware_calls_body(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         expected = [b"a", b""]
@@ -552,7 +552,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_body_in_app_after_middleware_calls_stream(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         assert await request.body() == b""
@@ -577,7 +577,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_body_in_app_after_middleware_calls_body(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         assert await request.body() == b"a"
@@ -599,7 +599,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_stream_in_dispatch_after_app_calls_stream(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         expected = [b"a", b""]
@@ -627,7 +627,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_stream_in_dispatch_after_app_calls_body(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         assert await request.body() == b"a"
@@ -705,7 +705,7 @@ async def send(msg: Message) -> None:
 
 
 def test_read_request_stream_in_dispatch_after_app_calls_body_with_middleware_calling_body_before_call_next(  # noqa: E501
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         assert await request.body() == b"a"
@@ -733,7 +733,7 @@ async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
 
 
 def test_read_request_body_in_dispatch_after_app_calls_body_with_middleware_calling_body_before_call_next(  # noqa: E501
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     async def homepage(request: Request):
         assert await request.body() == b"a"
@@ -837,7 +837,7 @@ async def send(msg: Message):
 
 
 def test_downstream_middleware_modifies_receive(
-    test_client_factory: Callable[[ASGIApp], TestClient]
+    test_client_factory: Callable[[ASGIApp], TestClient],
 ) -> None:
     """If a downstream middleware modifies receive() the final ASGI app
     should see the modified version.

tests/test_background.py

@@ -69,7 +69,7 @@ async def app(scope, receive, send):
 
 
 def test_multi_tasks_failure_avoids_next_execution(
-    test_client_factory: Callable[..., TestClient]
+    test_client_factory: Callable[..., TestClient],
 ) -> None:
     TASK_COUNTER = 0
 

tests/test_routing.py

@@ -1110,7 +1110,7 @@ async def modified_send(msg: Message) -> None:
 
 
 def test_websocket_route_middleware(
-    test_client_factory: typing.Callable[..., TestClient]
+    test_client_factory: typing.Callable[..., TestClient],
 ):
     async def websocket_endpoint(session: WebSocket):
         await session.accept()

tests/test_websockets.py

@@ -40,7 +40,7 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None:
 
 
 def test_websocket_ensure_unicode_on_send_json(
-    test_client_factory: Callable[..., TestClient]
+    test_client_factory: Callable[..., TestClient],
 ):
     async def app(scope: Scope, receive: Receive, send: Send) -> None:
         websocket = WebSocket(scope, receive=receive, send=send)