1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
|
from __future__ import annotations
import pdb # noqa: T100
from dataclasses import asdict, dataclass, field
from inspect import getmro
from sys import exc_info
from traceback import format_exception
from typing import TYPE_CHECKING, Any, Type, cast
from litestar.datastructures import Headers
from litestar.enums import MediaType, ScopeType
from litestar.exceptions import HTTPException, LitestarException, WebSocketException
from litestar.middleware.cors import CORSMiddleware
from litestar.middleware.exceptions._debug_response import _get_type_encoders_for_request, create_debug_response
from litestar.serialization import encode_json
from litestar.status_codes import HTTP_500_INTERNAL_SERVER_ERROR
from litestar.utils.deprecation import warn_deprecation
__all__ = ("ExceptionHandlerMiddleware", "ExceptionResponseContent", "create_exception_response")
if TYPE_CHECKING:
from starlette.exceptions import HTTPException as StarletteHTTPException
from litestar import Response
from litestar.app import Litestar
from litestar.connection import Request
from litestar.logging import BaseLoggingConfig
from litestar.types import (
ASGIApp,
ExceptionHandler,
ExceptionHandlersMap,
Logger,
Receive,
Scope,
Send,
)
from litestar.types.asgi_types import WebSocketCloseEvent
def get_exception_handler(exception_handlers: ExceptionHandlersMap, exc: Exception) -> ExceptionHandler | None:
"""Given a dictionary that maps exceptions and status codes to handler functions, and an exception, returns the
appropriate handler if existing.
Status codes are given preference over exception type.
If no status code match exists, each class in the MRO of the exception type is checked and
the first matching handler is returned.
Finally, if a ``500`` handler is registered, it will be returned for any exception that isn't a
subclass of :class:`HTTPException <litestar.exceptions.HTTPException>`.
Args:
exception_handlers: Mapping of status codes and exception types to handlers.
exc: Exception Instance to be resolved to a handler.
Returns:
Optional exception handler callable.
"""
if not exception_handlers:
return None
default_handler: ExceptionHandler | None = None
if isinstance(exc, HTTPException):
if exception_handler := exception_handlers.get(exc.status_code):
return exception_handler
else:
default_handler = exception_handlers.get(HTTP_500_INTERNAL_SERVER_ERROR)
return next(
(exception_handlers[cast("Type[Exception]", cls)] for cls in getmro(type(exc)) if cls in exception_handlers),
default_handler,
)
@dataclass
class ExceptionResponseContent:
"""Represent the contents of an exception-response."""
status_code: int
"""Exception status code."""
detail: str
"""Exception details or message."""
media_type: MediaType | str
"""Media type of the response."""
headers: dict[str, str] | None = field(default=None)
"""Headers to attach to the response."""
extra: dict[str, Any] | list[Any] | None = field(default=None)
"""An extra mapping to attach to the exception."""
def to_response(self, request: Request | None = None) -> Response:
"""Create a response from the model attributes.
Returns:
A response instance.
"""
from litestar.response import Response
content: Any = {k: v for k, v in asdict(self).items() if k not in ("headers", "media_type") and v is not None}
if self.media_type != MediaType.JSON:
content = encode_json(content)
return Response(
content=content,
headers=self.headers,
status_code=self.status_code,
media_type=self.media_type,
type_encoders=_get_type_encoders_for_request(request) if request is not None else None,
)
def _starlette_exception_handler(request: Request[Any, Any, Any], exc: StarletteHTTPException) -> Response:
return create_exception_response(
request=request,
exc=HTTPException(
detail=exc.detail,
status_code=exc.status_code,
headers=exc.headers,
),
)
def create_exception_response(request: Request[Any, Any, Any], exc: Exception) -> Response:
"""Construct a response from an exception.
Notes:
- For instances of :class:`HTTPException <litestar.exceptions.HTTPException>` or other exception classes that have a
``status_code`` attribute (e.g. Starlette exceptions), the status code is drawn from the exception, otherwise
response status is ``HTTP_500_INTERNAL_SERVER_ERROR``.
Args:
request: The request that triggered the exception.
exc: An exception.
Returns:
Response: HTTP response constructed from exception details.
"""
headers: dict[str, Any] | None
extra: dict[str, Any] | list | None
if isinstance(exc, HTTPException):
status_code = exc.status_code
headers = exc.headers
extra = exc.extra
else:
status_code = HTTP_500_INTERNAL_SERVER_ERROR
headers = None
extra = None
detail = (
exc.detail
if isinstance(exc, LitestarException) and status_code != HTTP_500_INTERNAL_SERVER_ERROR
else "Internal Server Error"
)
try:
media_type = request.route_handler.media_type
except (KeyError, AttributeError):
media_type = MediaType.JSON
content = ExceptionResponseContent(
status_code=status_code,
detail=detail,
headers=headers,
extra=extra,
media_type=media_type,
)
return content.to_response(request=request)
class ExceptionHandlerMiddleware:
"""Middleware used to wrap an ASGIApp inside a try catch block and handle any exceptions raised.
This used in multiple layers of Litestar.
"""
def __init__(self, app: ASGIApp, debug: bool | None, exception_handlers: ExceptionHandlersMap) -> None:
"""Initialize ``ExceptionHandlerMiddleware``.
Args:
app: The ``next`` ASGI app to call.
debug: Whether ``debug`` mode is enabled. Deprecated. Debug mode will be inferred from the request scope
exception_handlers: A dictionary mapping status codes and/or exception types to handler functions.
.. deprecated:: 2.0.0
The ``debug`` parameter is deprecated. It will be inferred from the request scope
"""
self.app = app
self.exception_handlers = exception_handlers
self.debug = debug
if debug is not None:
warn_deprecation(
"2.0.0",
deprecated_name="debug",
kind="parameter",
info="Debug mode will be inferred from the request scope",
)
self._get_debug = self._get_debug_scope if debug is None else lambda *a: debug
@staticmethod
def _get_debug_scope(scope: Scope) -> bool:
return scope["app"].debug
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""ASGI-callable.
Args:
scope: The ASGI connection scope.
receive: The ASGI receive function.
send: The ASGI send function.
Returns:
None
"""
try:
await self.app(scope, receive, send)
except Exception as e: # noqa: BLE001
litestar_app = scope["app"]
if litestar_app.logging_config and (logger := litestar_app.logger):
self.handle_exception_logging(logger=logger, logging_config=litestar_app.logging_config, scope=scope)
for hook in litestar_app.after_exception:
await hook(e, scope)
if litestar_app.pdb_on_exception:
pdb.post_mortem()
if scope["type"] == ScopeType.HTTP:
await self.handle_request_exception(
litestar_app=litestar_app, scope=scope, receive=receive, send=send, exc=e
)
else:
await self.handle_websocket_exception(send=send, exc=e)
async def handle_request_exception(
self, litestar_app: Litestar, scope: Scope, receive: Receive, send: Send, exc: Exception
) -> None:
"""Handle exception raised inside 'http' scope routes.
Args:
litestar_app: The litestar app instance.
scope: The ASGI connection scope.
receive: The ASGI receive function.
send: The ASGI send function.
exc: The caught exception.
Returns:
None.
"""
headers = Headers.from_scope(scope=scope)
if litestar_app.cors_config and (origin := headers.get("origin")):
cors_middleware = CORSMiddleware(app=self.app, config=litestar_app.cors_config)
send = cors_middleware.send_wrapper(send=send, origin=origin, has_cookie="cookie" in headers)
exception_handler = get_exception_handler(self.exception_handlers, exc) or self.default_http_exception_handler
request: Request[Any, Any, Any] = litestar_app.request_class(scope=scope, receive=receive, send=send)
response = exception_handler(request, exc)
await response.to_asgi_response(app=None, request=request)(scope=scope, receive=receive, send=send)
@staticmethod
async def handle_websocket_exception(send: Send, exc: Exception) -> None:
"""Handle exception raised inside 'websocket' scope routes.
Args:
send: The ASGI send function.
exc: The caught exception.
Returns:
None.
"""
code = 4000 + HTTP_500_INTERNAL_SERVER_ERROR
reason = "Internal Server Error"
if isinstance(exc, WebSocketException):
code = exc.code
reason = exc.detail
elif isinstance(exc, LitestarException):
reason = exc.detail
event: WebSocketCloseEvent = {"type": "websocket.close", "code": code, "reason": reason}
await send(event)
def default_http_exception_handler(self, request: Request, exc: Exception) -> Response[Any]:
"""Handle an HTTP exception by returning the appropriate response.
Args:
request: An HTTP Request instance.
exc: The caught exception.
Returns:
An HTTP response.
"""
status_code = exc.status_code if isinstance(exc, HTTPException) else HTTP_500_INTERNAL_SERVER_ERROR
if status_code == HTTP_500_INTERNAL_SERVER_ERROR and self._get_debug_scope(request.scope):
return create_debug_response(request=request, exc=exc)
return create_exception_response(request=request, exc=exc)
def handle_exception_logging(self, logger: Logger, logging_config: BaseLoggingConfig, scope: Scope) -> None:
"""Handle logging - if the litestar app has a logging config in place.
Args:
logger: A logger instance.
logging_config: Logging Config instance.
scope: The ASGI connection scope.
Returns:
None
"""
if (
logging_config.log_exceptions == "always"
or (logging_config.log_exceptions == "debug" and self._get_debug_scope(scope))
) and logging_config.exception_logging_handler:
logging_config.exception_logging_handler(logger, scope, format_exception(*exc_info()))
|