0
# Exception Handling
1
2
Framework-specific exceptions and error handling including HTTP exceptions, validation errors, and custom exception handlers. Litestar provides structured error responses and comprehensive exception handling with debugging support.
3
4
## Capabilities
5
6
### Base Exceptions
7
8
Core exception classes that serve as the foundation for all framework exceptions.
9
10
```python { .api }
11
class LitestarException(Exception):
12
"""Base exception for all Litestar-specific errors."""
13
14
def __init__(self, detail: str = "", *args: object) -> None:
15
"""
16
Initialize base Litestar exception.
17
18
Parameters:
19
- detail: Human-readable error description
20
- *args: Additional exception arguments
21
"""
22
super().__init__(detail, *args)
23
self.detail = detail
24
25
class LitestarWarning(UserWarning):
26
"""Base warning class for Litestar framework warnings."""
27
28
class MissingDependencyException(LitestarException, ImportError):
29
"""Exception raised when a required dependency is not available."""
30
31
class SerializationException(LitestarException):
32
"""Exception raised during data serialization/deserialization."""
33
34
class ImproperlyConfiguredException(LitestarException):
35
"""Exception raised when the application is improperly configured."""
36
```
37
38
### HTTP Exceptions
39
40
HTTP-specific exceptions that generate appropriate HTTP error responses.
41
42
```python { .api }
43
class HTTPException(LitestarException):
44
def __init__(
45
self,
46
detail: str = "",
47
*,
48
status_code: int = 500,
49
headers: dict[str, str] | None = None,
50
extra: dict[str, Any] | list[Any] | None = None,
51
):
52
"""
53
Base HTTP exception.
54
55
Parameters:
56
- detail: Human-readable error description
57
- status_code: HTTP status code
58
- headers: Additional response headers
59
- extra: Additional data to include in error response
60
"""
61
super().__init__(detail)
62
self.status_code = status_code
63
self.headers = headers or {}
64
self.extra = extra
65
66
def to_response(self) -> Response:
67
"""Convert exception to HTTP response."""
68
69
class ClientException(HTTPException):
70
"""Base class for 4xx client error exceptions."""
71
status_code = 400
72
73
class InternalServerException(HTTPException):
74
"""Base class for 5xx server error exceptions."""
75
status_code = 500
76
77
# 4xx Client Error Exceptions
78
class BadRequestException(ClientException):
79
"""400 Bad Request exception."""
80
status_code = HTTP_400_BAD_REQUEST
81
82
class NotAuthorizedException(ClientException):
83
"""401 Unauthorized exception."""
84
status_code = HTTP_401_UNAUTHORIZED
85
86
class PermissionDeniedException(ClientException):
87
"""403 Forbidden exception."""
88
status_code = HTTP_403_FORBIDDEN
89
90
class NotFoundException(ClientException):
91
"""404 Not Found exception."""
92
status_code = HTTP_404_NOT_FOUND
93
94
class MethodNotAllowedException(ClientException):
95
"""405 Method Not Allowed exception."""
96
status_code = HTTP_405_METHOD_NOT_ALLOWED
97
98
def __init__(
99
self,
100
detail: str = "",
101
*,
102
method: str,
103
path: str,
104
**kwargs: Any,
105
):
106
"""
107
Method not allowed exception.
108
109
Parameters:
110
- detail: Error description
111
- method: HTTP method that was not allowed
112
- path: Request path
113
"""
114
super().__init__(detail, **kwargs)
115
self.method = method
116
self.path = path
117
118
class NotAcceptableException(ClientException):
119
"""406 Not Acceptable exception."""
120
status_code = HTTP_406_NOT_ACCEPTABLE
121
122
class RequestTimeoutException(ClientException):
123
"""408 Request Timeout exception."""
124
status_code = HTTP_408_REQUEST_TIMEOUT
125
126
class ConflictException(ClientException):
127
"""409 Conflict exception."""
128
status_code = HTTP_409_CONFLICT
129
130
class ValidationException(ClientException):
131
"""422 Unprocessable Entity exception for validation errors."""
132
status_code = HTTP_422_UNPROCESSABLE_ENTITY
133
134
class TooManyRequestsException(ClientException):
135
"""429 Too Many Requests exception."""
136
status_code = HTTP_429_TOO_MANY_REQUESTS
137
138
# 5xx Server Error Exceptions
139
class InternalServerException(InternalServerException):
140
"""500 Internal Server Error exception."""
141
status_code = HTTP_500_INTERNAL_SERVER_ERROR
142
143
class NotImplementedException(InternalServerException):
144
"""501 Not Implemented exception."""
145
status_code = HTTP_501_NOT_IMPLEMENTED
146
147
class BadGatewayException(InternalServerException):
148
"""502 Bad Gateway exception."""
149
status_code = HTTP_502_BAD_GATEWAY
150
151
class ServiceUnavailableException(InternalServerException):
152
"""503 Service Unavailable exception."""
153
status_code = HTTP_503_SERVICE_UNAVAILABLE
154
```
155
156
### WebSocket Exceptions
157
158
Exceptions specific to WebSocket connections.
159
160
```python { .api }
161
class WebSocketException(LitestarException):
162
def __init__(self, detail: str = "", code: int = 1011):
163
"""
164
WebSocket exception.
165
166
Parameters:
167
- detail: Error description
168
- code: WebSocket close code
169
"""
170
super().__init__(detail)
171
self.code = code
172
173
class WebSocketDisconnect(WebSocketException):
174
"""Exception raised when WebSocket connection is unexpectedly closed."""
175
176
def __init__(self, code: int = 1000, reason: str | None = None):
177
"""
178
WebSocket disconnect exception.
179
180
Parameters:
181
- code: WebSocket close code
182
- reason: Disconnect reason
183
"""
184
detail = f"WebSocket disconnected (code: {code})"
185
if reason:
186
detail += f" - {reason}"
187
super().__init__(detail, code)
188
self.reason = reason
189
```
190
191
### DTO Exceptions
192
193
Exceptions related to Data Transfer Object operations.
194
195
```python { .api }
196
class DTOFactoryException(LitestarException):
197
"""Exception raised during DTO factory operations."""
198
199
class InvalidAnnotationException(DTOFactoryException):
200
"""Exception raised when encountering invalid type annotations."""
201
```
202
203
### Route and Template Exceptions
204
205
Exceptions related to routing and template processing.
206
207
```python { .api }
208
class NoRouteMatchFoundException(InternalServerException):
209
"""Exception raised when no route matches the request."""
210
status_code = HTTP_404_NOT_FOUND
211
212
class TemplateNotFoundException(InternalServerException):
213
"""Exception raised when a template cannot be found."""
214
status_code = HTTP_500_INTERNAL_SERVER_ERROR
215
```
216
217
### Exception Handlers
218
219
Functions and classes for handling exceptions and generating error responses.
220
221
```python { .api }
222
def create_exception_response(
223
request: Request,
224
exc: Exception,
225
) -> Response:
226
"""
227
Create an HTTP response from an exception.
228
229
Parameters:
230
- request: HTTP request that caused the exception
231
- exc: Exception to convert to response
232
233
Returns:
234
HTTP response representing the exception
235
"""
236
237
class ExceptionHandler:
238
def __init__(
239
self,
240
handler: Callable[[Request, Exception], Response | Awaitable[Response]],
241
*,
242
status_code: int | None = None,
243
media_type: MediaType | str | None = None,
244
):
245
"""
246
Exception handler configuration.
247
248
Parameters:
249
- handler: Function to handle the exception
250
- status_code: Override status code
251
- media_type: Response media type
252
"""
253
254
def exception_handler(
255
exc_class: type[Exception],
256
*,
257
status_code: int | None = None,
258
media_type: MediaType | str | None = None,
259
) -> Callable[[T], T]:
260
"""
261
Decorator to register exception handlers.
262
263
Parameters:
264
- exc_class: Exception class to handle
265
- status_code: Override status code
266
- media_type: Response media type
267
268
Returns:
269
Decorator function
270
"""
271
```
272
273
### Validation Error Details
274
275
Structured validation error information for detailed error responses.
276
277
```python { .api }
278
class ValidationErrorDetail:
279
def __init__(
280
self,
281
message: str,
282
key: str,
283
source: Literal["body", "query", "path", "header", "cookie"] | None = None,
284
):
285
"""
286
Validation error detail.
287
288
Parameters:
289
- message: Error message
290
- key: Field or parameter name that failed validation
291
- source: Source of the invalid data
292
"""
293
self.message = message
294
self.key = key
295
self.source = source
296
297
def to_dict(self) -> dict[str, Any]:
298
"""Convert to dictionary representation."""
299
return {
300
"message": self.message,
301
"key": self.key,
302
"source": self.source,
303
}
304
```
305
306
## Usage Examples
307
308
### Basic Exception Handling
309
310
```python
311
from litestar import get
312
from litestar.exceptions import HTTPException, NotFoundException
313
from litestar.status_codes import HTTP_400_BAD_REQUEST
314
315
@get("/users/{user_id:int}")
316
def get_user(user_id: int) -> dict:
317
if user_id < 1:
318
raise HTTPException(
319
detail="User ID must be positive",
320
status_code=HTTP_400_BAD_REQUEST,
321
extra={"provided_id": user_id}
322
)
323
324
# Simulate user lookup
325
if user_id == 999:
326
raise NotFoundException("User not found")
327
328
return {"id": user_id, "name": "Alice"}
329
330
@get("/protected")
331
def protected_resource() -> dict:
332
# Check authentication (simplified)
333
authenticated = False
334
if not authenticated:
335
raise NotAuthorizedException("Authentication required")
336
337
return {"data": "secret information"}
338
```
339
340
### Custom Exception Classes
341
342
```python
343
from litestar.exceptions import HTTPException
344
from litestar.status_codes import HTTP_402_PAYMENT_REQUIRED
345
346
class InsufficientCreditsException(HTTPException):
347
"""Custom exception for payment/credits system."""
348
status_code = HTTP_402_PAYMENT_REQUIRED
349
350
def __init__(self, required_credits: int, available_credits: int):
351
detail = f"Insufficient credits: need {required_credits}, have {available_credits}"
352
super().__init__(
353
detail=detail,
354
extra={
355
"required_credits": required_credits,
356
"available_credits": available_credits,
357
"credit_purchase_url": "/purchase-credits"
358
}
359
)
360
361
class RateLimitExceededException(HTTPException):
362
"""Custom exception for rate limiting."""
363
status_code = HTTP_429_TOO_MANY_REQUESTS
364
365
def __init__(self, limit: int, window: int, retry_after: int):
366
detail = f"Rate limit exceeded: {limit} requests per {window} seconds"
367
super().__init__(
368
detail=detail,
369
headers={"Retry-After": str(retry_after)},
370
extra={
371
"limit": limit,
372
"window": window,
373
"retry_after": retry_after
374
}
375
)
376
377
@get("/premium-feature")
378
def premium_feature(user_credits: int = 10) -> dict:
379
required_credits = 50
380
381
if user_credits < required_credits:
382
raise InsufficientCreditsException(required_credits, user_credits)
383
384
return {"result": "premium feature accessed"}
385
```
386
387
### Global Exception Handlers
388
389
```python
390
from litestar import Litestar
391
from litestar.exceptions import HTTPException, ValidationException
392
from litestar.response import Response
393
import logging
394
395
logger = logging.getLogger(__name__)
396
397
def generic_exception_handler(request: Request, exc: Exception) -> Response:
398
"""Handle unexpected exceptions."""
399
logger.exception("Unhandled exception occurred")
400
401
return Response(
402
content={
403
"error": "Internal server error",
404
"detail": "An unexpected error occurred",
405
"request_id": getattr(request.state, "request_id", None)
406
},
407
status_code=500,
408
headers={"X-Error-Type": "UnhandledException"}
409
)
410
411
def validation_exception_handler(request: Request, exc: ValidationException) -> Response:
412
"""Handle validation errors with detailed information."""
413
return Response(
414
content={
415
"error": "Validation failed",
416
"detail": exc.detail,
417
"validation_errors": exc.extra or [],
418
"path": str(request.url.path),
419
"method": request.method
420
},
421
status_code=exc.status_code,
422
headers={"X-Error-Type": "ValidationError"}
423
)
424
425
def http_exception_handler(request: Request, exc: HTTPException) -> Response:
426
"""Handle HTTP exceptions with consistent format."""
427
return Response(
428
content={
429
"error": exc.__class__.__name__,
430
"detail": exc.detail,
431
"status_code": exc.status_code,
432
**({"extra": exc.extra} if exc.extra else {})
433
},
434
status_code=exc.status_code,
435
headers=exc.headers
436
)
437
438
app = Litestar(
439
route_handlers=[...],
440
exception_handlers={
441
HTTPException: http_exception_handler,
442
ValidationException: validation_exception_handler,
443
Exception: generic_exception_handler,
444
}
445
)
446
```
447
448
### Validation Exception with Details
449
450
```python
451
from litestar.exceptions import ValidationException, ValidationErrorDetail
452
453
@post("/users")
454
def create_user(data: dict) -> dict:
455
errors = []
456
457
# Validate required fields
458
if not data.get("name"):
459
errors.append(ValidationErrorDetail(
460
message="Name is required",
461
key="name",
462
source="body"
463
))
464
465
if not data.get("email"):
466
errors.append(ValidationErrorDetail(
467
message="Email is required",
468
key="email",
469
source="body"
470
))
471
elif "@" not in data["email"]:
472
errors.append(ValidationErrorDetail(
473
message="Invalid email format",
474
key="email",
475
source="body"
476
))
477
478
if errors:
479
raise ValidationException(
480
detail="User validation failed",
481
extra=[error.to_dict() for error in errors]
482
)
483
484
return {"id": 123, "name": data["name"], "email": data["email"]}
485
```
486
487
### WebSocket Exception Handling
488
489
```python
490
from litestar import websocket, WebSocket
491
from litestar.exceptions import WebSocketException, WebSocketDisconnect
492
493
@websocket("/ws")
494
async def websocket_handler(websocket: WebSocket) -> None:
495
try:
496
await websocket.accept()
497
498
while True:
499
message = await websocket.receive_json()
500
501
# Validate message format
502
if not isinstance(message, dict) or "type" not in message:
503
raise WebSocketException(
504
"Invalid message format: must be JSON object with 'type' field",
505
code=1003 # Unsupported data
506
)
507
508
# Process message
509
await websocket.send_json({"status": "received", "data": message})
510
511
except WebSocketDisconnect as exc:
512
logger.info(f"WebSocket disconnected: {exc.reason}")
513
except WebSocketException as exc:
514
logger.error(f"WebSocket error: {exc.detail}")
515
await websocket.close(code=exc.code, reason=exc.detail)
516
except Exception as exc:
517
logger.exception("Unexpected error in WebSocket handler")
518
await websocket.close(code=1011, reason="Internal error")
519
```
520
521
### Exception Handler Decorator
522
523
```python
524
from litestar.exceptions import exception_handler
525
526
@exception_handler(ValueError)
527
def handle_value_error(request: Request, exc: ValueError) -> Response:
528
"""Handle ValueError exceptions."""
529
return Response(
530
content={
531
"error": "Invalid value",
532
"detail": str(exc),
533
"path": str(request.url.path)
534
},
535
status_code=400
536
)
537
538
@exception_handler(KeyError)
539
def handle_key_error(request: Request, exc: KeyError) -> Response:
540
"""Handle KeyError exceptions."""
541
return Response(
542
content={
543
"error": "Missing required field",
544
"detail": f"Required field not found: {exc}",
545
"path": str(request.url.path)
546
},
547
status_code=400
548
)
549
550
# Apply to specific routes
551
@get("/data/{key:str}", exception_handlers={KeyError: handle_key_error})
552
def get_data(key: str) -> dict:
553
data = {"user": "alice", "admin": "bob"}
554
return {"value": data[key]} # May raise KeyError
555
```
556
557
### Error Response Formatting
558
559
```python
560
from litestar.response import Response
561
from litestar.status_codes import *
562
563
class APIErrorResponse:
564
"""Standardized API error response format."""
565
566
def __init__(
567
self,
568
error_code: str,
569
message: str,
570
details: dict[str, Any] | None = None,
571
status_code: int = 500,
572
):
573
self.error_code = error_code
574
self.message = message
575
self.details = details or {}
576
self.status_code = status_code
577
578
def to_response(self) -> Response:
579
content = {
580
"success": False,
581
"error": {
582
"code": self.error_code,
583
"message": self.message,
584
"timestamp": datetime.utcnow().isoformat(),
585
**self.details
586
}
587
}
588
return Response(content=content, status_code=self.status_code)
589
590
def api_exception_handler(request: Request, exc: HTTPException) -> Response:
591
"""Convert HTTP exceptions to standardized API error format."""
592
error_mapping = {
593
400: "BAD_REQUEST",
594
401: "UNAUTHORIZED",
595
403: "FORBIDDEN",
596
404: "NOT_FOUND",
597
422: "VALIDATION_ERROR",
598
429: "RATE_LIMIT_EXCEEDED",
599
500: "INTERNAL_ERROR",
600
}
601
602
error_code = error_mapping.get(exc.status_code, "UNKNOWN_ERROR")
603
604
api_error = APIErrorResponse(
605
error_code=error_code,
606
message=exc.detail,
607
details=exc.extra or {},
608
status_code=exc.status_code
609
)
610
611
return api_error.to_response()
612
```
613
614
### Debugging and Development Exception Handling
615
616
```python
617
import traceback
618
from litestar import Litestar
619
620
def debug_exception_handler(request: Request, exc: Exception) -> Response:
621
"""Detailed exception handler for development."""
622
return Response(
623
content={
624
"error": exc.__class__.__name__,
625
"message": str(exc),
626
"traceback": traceback.format_exc().split('\n'),
627
"request": {
628
"method": request.method,
629
"url": str(request.url),
630
"headers": dict(request.headers),
631
"path_params": request.path_params,
632
"query_params": dict(request.query_params),
633
}
634
},
635
status_code=500,
636
headers={"Content-Type": "application/json"}
637
)
638
639
def create_app(debug: bool = False) -> Litestar:
640
exception_handlers = {}
641
642
if debug:
643
exception_handlers[Exception] = debug_exception_handler
644
else:
645
exception_handlers[Exception] = generic_exception_handler
646
647
return Litestar(
648
route_handlers=[...],
649
exception_handlers=exception_handlers,
650
debug=debug
651
)
652
```
653
654
## Types
655
656
```python { .api }
657
# Exception handler types
658
ExceptionHandler = Callable[[Request, Exception], Response | Awaitable[Response]]
659
ExceptionHandlersMap = dict[type[Exception], ExceptionHandler]
660
661
# Validation error types
662
ValidationSource = Literal["body", "query", "path", "header", "cookie"]
663
664
# WebSocket close codes (RFC 6455)
665
WS_CLOSE_CODES = {
666
1000: "Normal Closure",
667
1001: "Going Away",
668
1002: "Protocol Error",
669
1003: "Unsupported Data",
670
1007: "Invalid Frame Payload Data",
671
1008: "Policy Violation",
672
1009: "Message Too Big",
673
1010: "Mandatory Extension",
674
1011: "Internal Error",
675
1015: "TLS Handshake",
676
}
677
678
# Custom application close codes (4000-4999)
679
WS_CUSTOM_CLOSE_CODES = {
680
4001: "Unauthorized",
681
4003: "Forbidden",
682
4004: "Not Found",
683
4008: "Rate Limited",
684
4009: "Invalid Request",
685
}
686
```