Allow async exception handlers to type-check

starlette/_exception_handler.py

@@ -58,7 +58,7 @@ async def sender(message: Message) -> None:
             if is_async_callable(handler):
                 response = await handler(conn, exc)
             else:
-                response = await run_in_threadpool(handler, conn, exc)  # type: ignore
+                response = await run_in_threadpool(handler, conn, exc)
             if response is not None:
                 await response(scope, receive, sender)
 

starlette/_utils.py

@@ -9,10 +9,10 @@
 
 from starlette.types import Scope
 
-if sys.version_info >= (3, 10):  # pragma: no cover
-    from typing import TypeGuard
+if sys.version_info >= (3, 13):  # pragma: no cover
+    from typing import TypeIs
 else:  # pragma: no cover
-    from typing_extensions import TypeGuard
+    from typing_extensions import TypeIs
 
 has_exceptiongroups = True
 if sys.version_info < (3, 11):  # pragma: no cover
@@ -26,11 +26,11 @@
 
 
 @overload
-def is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]: ...
+def is_async_callable(obj: AwaitableCallable[T]) -> TypeIs[AwaitableCallable[T]]: ...
 
 
 @overload
-def is_async_callable(obj: Any) -> TypeGuard[AwaitableCallable[Any]]: ...
+def is_async_callable(obj: Any) -> TypeIs[AwaitableCallable[Any]]: ...
 
 
 def is_async_callable(obj: Any) -> Any:

starlette/applications.py

@@ -80,7 +80,7 @@ def __init__(
     def build_middleware_stack(self) -> ASGIApp:
         debug = self.debug
         error_handler = None
-        exception_handlers: dict[Any, Callable[[Request, Exception], Response]] = {}
+        exception_handlers: dict[Any, ExceptionHandler] = {}
 
         for key, value in self.exception_handlers.items():
             if key in (500, Exception):

starlette/middleware/errors.py

@@ -4,13 +4,12 @@
 import inspect
 import sys
 import traceback
-from typing import Any, Callable
 
 from starlette._utils import is_async_callable
 from starlette.concurrency import run_in_threadpool
 from starlette.requests import Request
 from starlette.responses import HTMLResponse, PlainTextResponse, Response
-from starlette.types import ASGIApp, Message, Receive, Scope, Send
+from starlette.types import ASGIApp, ExceptionHandler, Message, Receive, Scope, Send
 
 STYLES = """
 p {
@@ -140,7 +139,7 @@ class ServerErrorMiddleware:
     def __init__(
         self,
         app: ASGIApp,
-        handler: Callable[[Request, Exception], Any] | None = None,
+        handler: ExceptionHandler | None = None,
         debug: bool = False,
     ) -> None:
         self.app = app

starlette/middleware/exceptions.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Any, Callable
+from typing import Any
 
 from starlette._exception_handler import (
     ExceptionHandlers,
@@ -11,15 +11,15 @@
 from starlette.exceptions import HTTPException, WebSocketException
 from starlette.requests import Request
 from starlette.responses import PlainTextResponse, Response
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, ExceptionHandler, Receive, Scope, Send
 from starlette.websockets import WebSocket
 
 
 class ExceptionMiddleware:
     def __init__(
         self,
         app: ASGIApp,
-        handlers: Mapping[Any, Callable[[Request, Exception], Response]] | None = None,
+        handlers: Mapping[Any, ExceptionHandler] | None = None,
         debug: bool = False,
     ) -> None:
         self.app = app
@@ -36,7 +36,7 @@ def __init__(
     def add_exception_handler(
         self,
         exc_class_or_status_code: int | type[Exception],
-        handler: Callable[[Request, Exception], Response],
+        handler: ExceptionHandler,
     ) -> None:
         if isinstance(exc_class_or_status_code, int):
             self._status_handlers[exc_class_or_status_code] = handler

starlette/routing.py

@@ -65,7 +65,7 @@ def request_response(
     and returns an ASGI application.
     """
     f: Callable[[Request], Awaitable[Response]] = (
-        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore
+        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)
     )
 
     async def app(scope: Scope, receive: Receive, send: Send) -> None:

tests/test_exceptions.py

@@ -48,6 +48,14 @@ async def handler_that_reads_body(request: Request, exc: BadBodyException) -> JS
     return JSONResponse(status_code=422, content={"body": body.decode()})
 
 
+async def async_catch_all_handler(request: Request, exc: Exception) -> JSONResponse:
+    raise NotImplementedError
+
+
+def sync_catch_all_handler(request: Request, exc: Exception) -> JSONResponse:
+    raise NotImplementedError
+
+
 class HandledExcAfterResponse:
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         response = PlainTextResponse("OK", status_code=200)
@@ -74,6 +82,9 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
     handlers={BadBodyException: handler_that_reads_body},  # type: ignore[dict-item]
 )
 
+sync_catch_all = ExceptionMiddleware(router, handlers={Exception: sync_catch_all_handler})
+async_catch_all = ExceptionMiddleware(router, handlers={Exception: async_catch_all_handler})
+
 
 @pytest.fixture
 def client(test_client_factory: TestClientFactory) -> Generator[TestClient, None, None]: