resolved RESOLVED UNSUBMITTED PASS · None tool calls · 0 s · fastapi/fastapi
🚸 Improve tracebacks by adding endpoint metadata
Today, when validation errors occur, error messages don't indicate what endpoint caused the problem, which can make debugging difficult.
This PR adds endpoint context metadata to validation error messages to show the filepath (clickable in IDEs!) and line number, function name and HTTP method and route where applicable. The endpoint context is extracted once per endpoint and then cached to avoid any additional performance overhead on subsequent requests. Just to double check, I also ran a rough benchmark (100k requests):
```
Baseline (without changes): 0.0006s
Uncached: 1.3281s (13.28µs per request)
**Cached**: 0.0060s (0.06µs per request)
```
Endpoint context only appears in the exception's string representation (server logs). It is **not** exposed to clients in HTTP responses. Clients still receive the standard 422 response with validation error details.
## Examples
### ResponseValidationError
**Before:**
```
fastapi.exceptions.ResponseValidationError: 1 validation error:
{'type': 'missing', 'loc': ('response', 'id'), 'msg': 'Field required', 'input': {'name': 'Widget'}}
```
**After:**
```
fastapi.exceptions.ResponseValidationError: 1 validation error:
{'type': 'missing', 'loc': ('response', 'id'), 'msg': 'Field required', 'input': {'name': 'Widget'}}
File "/Users/savannah/work/fastapi/capture_real_examples.py", line 43, in create_item
POST /items/
```
### RequestValidationError
**Before:**
```
fastapi.exceptions.RequestValidationError: 1 validation error:
{'type': 'int_parsing', 'loc': ('path', 'user_id'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'invalid'}
```
**After:**
```
fastapi.exceptions.RequestValidationError: 1 validation error:
{'type': 'int_parsing', 'loc': ('path', 'user_id'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'invalid'}
File "/Users/savannah/work/fastapi/capture_real_examples.py", line 38, in get_user
GET /users/{user_id}
```
### WebSocketRequestValidationError
**Before:**
```
fastapi.exceptions.WebSocketRequestValidationError: 1 validation error:
{'type': 'int_parsing', 'loc': ('path', 'item_id'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'invalid'}
```
**After:**
```
fastapi.exceptions.WebSocketRequestValidationError: 1 validation error:
{'type': 'int_parsing', 'loc': ('path', 'item_id'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'invalid'}
File "/Users/savannah/work/fastapi/capture_real_examples.py", line 48, in websocket_endpoint
WS /ws/{item_id}
```| # | Tool | Arguments | Result |
|---|---|---|---|
| No trace captured. | |||
--- a/fastapi/exceptions.py
+++ b/fastapi/exceptions.py
@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional, Sequence, Type, Union
+from typing import Any, Dict, Optional, Sequence, Type, TypedDict, Union
from annotated_doc import Doc
from pydantic import BaseModel, create_model
@@ -7,6 +7,13 @@
from typing_extensions import Annotated
+class EndpointContext(TypedDict, total=False):
+ function: str
+ path: str
+ file: str
+ line: int
+
+
class HTTPException(StarletteHTTPException):
"""
An HTTP exception you can raise in your own code to show errors to the client.
@@ -155,30 +162,72 @@ class DependencyScopeError(FastAPIError):
class ValidationException(Exception):
- def __init__(self, errors: Sequence[Any]) -> None:
+ def __init__(
+ self,
+ errors: Sequence[Any],
+ *,
+ endpoint_ctx: Optional[EndpointContext] = None,
+ ) -> None:
self._errors = errors
+ self.endpoint_ctx = endpoint_ctx
+
+ ctx = endpoint_ctx or {}
+ self.endpoint_function = ctx.get("function")
+ self.endpoint_path = ctx.get("path")
+ self.endpoint_file = ctx.get("file")
+ self.endpoint_line = ctx.get("line")
def errors(self) -> Sequence[Any]:
return self._errors
+ def _format_endpoint_context(self) -> str:
+ if not (self.endpoint_file and self.endpoint_line and self.endpoint_function):
+ if self.endpoint_path:
+ return f"\n Endpoint: {self.endpoint_path}"
+ return ""
+
+ context = f'\n File "{self.endpoint_file}", line {self.endpoint_line}, in {self.endpoint_function}'
+ if self.endpoint_path:
+ context += f"\n {self.endpoint_path}"
+ return context
+
+ def __str__(self) -> str:
+ message = f"{len(self._errors)} validation error{'s' if len(self._errors) != 1 else ''}:\n"
+ for err in self._errors:
+ message += f" {err}\n"
+ message += self._format_endpoint_context()
+ return message.rstrip()
+
class RequestValidationError(ValidationException):
- def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
- super().__init__(errors)
+ def __init__(
+ self,
+ errors: Sequence[Any],
+ *,
+ body: Any = None,
+ endpoint_ctx: Optional[EndpointContext] = None,
+ ) -> None:
+ super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
class WebSocketRequestValidationError(ValidationException):
- pass
+ def __init__(
+ self,
+ errors: Sequence[Any],
+ *,
+ endpoint_ctx: Optional[EndpointContext] = None,
+ ) -> None:
+ super().__init__(errors, endpoint_ctx=endpoint_ctx)
class ResponseValidationError(ValidationException):
- def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:
- super().__init__(errors)
+ def __init__(
+ self,
+ errors: Sequence[Any],
+ *,
+ body: Any = None,
+ endpoint_ctx: Optional[EndpointContext] = None,
+ ) -> None:
+ super().__init__(errors, endpoint_ctx=endpoint_ctx)
self.body = body
-
- def __str__(self) -> str:
- message = f"{len(self._errors)} validation errors:\n"
- for err in self._errors:
- message += f" {err}\n"
- return message
--- a/fastapi/routing.py
+++ b/fastapi/routing.py
@@ -46,6 +46,7 @@
)
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import (
+ EndpointContext,
FastAPIError,
RequestValidationError,
ResponseValidationError,
@@ -212,6 +213,33 @@ async def merged_lifespan(
return merged_lifespan # type: ignore[return-value]
+# Cache for endpoint context to avoid re-extracting on every request
+_endpoint_context_cache: Dict[int, EndpointContext] = {}
+
+
+def _extract_endpoint_context(func: Any) -> EndpointContext:
+ """Extract endpoint context with caching to avoid repeated file I/O."""
+ func_id = id(func)
+
+ if func_id in _endpoint_context_cache:
+ return _endpoint_context_cache[func_id]
+
+ try:
+ ctx: EndpointContext = {}
+
+ if (source_file := inspect.getsourcefile(func)) is not None:
+ ctx["file"] = source_file
+ if (line_number := inspect.getsourcelines(func)[1]) is not None:
+ ctx["line"] = line_number
+ if (func_name := getattr(func, "__name__", None)) is not None:
+ ctx["function"] = func_name
+ except Exception:
+ ctx = EndpointContext()
+
+ _endpoint_context_cache[func_id] = ctx
+ return ctx
+
+
async def serialize_response(
*,
field: Optional[ModelField] = None,
@@ -223,6 +251,7 @@ async def serialize_response(
exclude_defaults: bool = False,
exclude_none: bool = False,
is_coroutine: bool = True,
+ endpoint_ctx: Optional[EndpointContext] = None,
) -> Any:
if field:
errors = []
@@ -245,8 +274,11 @@ async def serialize_response(
elif errors_:
errors.append(errors_)
if errors:
+ ctx = endpoint_ctx or EndpointContext()
raise ResponseValidationError(
- errors=_normalize_errors(errors), body=response_content
+ errors=_normalize_errors(errors),
+ body=response_content,
+ endpoint_ctx=ctx,
)
if hasattr(field, "serialize"):
@@ -318,6 +350,18 @@ async def app(request: Request) -> Response:
"fastapi_middleware_astack not found in request scope"
)
+ # Extract endpoint context for error messages
+ endpoint_ctx = (
+ _extract_endpoint_context(dependant.call)
+ if dependant.call
+ else EndpointContext()
+ )
+
+ if dependant.path:
+ # For mounted sub-apps, include the mount path prefix
+ mount_path = request.scope.get("root_path", "").rstrip("/")
+ endpoint_ctx["path"] = f"{request.method} {mount_path}{dependant.path}"
+
# Read body and auto-close files
try:
body: Any = None
@@ -355,6 +399,7 @@ async def app(request: Request) -> Response:
}
],
body=e.doc,
+ endpoint_ctx=endpoint_ctx,
)
raise validation_error from e
except HTTPException:
@@ -414,14 +459,15 @@ async def app(request: Request) -> Response:
exclude_defaults=response_model_exclude_defaults,
exclude_none=response_model_exclude_none,
is_coroutine=is_coroutine,
+ endpoint_ctx=endpoint_ctx,
)
response = actual_response_class(content, **response_args)
if not is_body_allowed_for_status_code(response.status_code):
response.body = b""
response.headers.raw.extend(solved_result.response.headers.raw)
if errors:
validation_error = RequestValidationError(
- _normalize_errors(errors), body=body
+ _normalize_errors(errors), body=body, endpoint_ctx=endpoint_ctx
)
raise validation_error
@@ -438,6 +484,15 @@ def get_websocket_app(
embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
async def app(websocket: WebSocket) -> None:
+ endpoint_ctx = (
+ _extract_endpoint_context(dependant.call)
+ if dependant.call
+ else EndpointContext()
+ )
+ if dependant.path:
+ # For mounted sub-apps, include the mount path prefix
+ mount_path = websocket.scope.get("root_path", "").rstrip("/")
+ endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}"
async_exit_stack = websocket.scope.get("fastapi_inner_astack")
assert isinstance(async_exit_stack, AsyncExitStack), (
"fastapi_inner_astack not found in request scope"
@@ -451,7 +506,8 @@ async def app(websocket: WebSocket) -> None:
)
if solved_result.errors:
raise WebSocketRequestValidationError(
- _normalize_errors(solved_result.errors)
+ _normalize_errors(solved_result.errors),
+ endpoint_ctx=endpoint_ctx,
)
assert dependant.call is not None, "dependant.call must be a function"
await dependant.call(**solved_result.values)....... [100%]
=============================== warnings summary ===============================
../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-0.50.0-py3-none-any/starlette/testclient.py:45
/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-0.50.0-py3-none-any/starlette/testclient.py:45: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
_PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
7 passed, 1 warning in 0.44s