0
# Exception Handling & Status Codes
1
2
Starlette provides comprehensive exception handling with HTTP and WebSocket exceptions, complete status code constants, and flexible error handling mechanisms for building robust web applications.
3
4
## HTTP Exceptions
5
6
### HTTPException Class
7
8
```python { .api }
9
from starlette.exceptions import HTTPException, WebSocketException
10
from starlette.responses import Response
11
from collections.abc import Mapping
12
from typing import Any
13
14
class HTTPException(Exception):
15
"""
16
HTTP error exception with status code and optional details.
17
18
Used to raise HTTP errors that will be converted to appropriate
19
HTTP responses by exception middleware.
20
"""
21
22
def __init__(
23
self,
24
status_code: int,
25
detail: str | None = None,
26
headers: Mapping[str, str] | None = None,
27
) -> None:
28
"""
29
Initialize HTTP exception.
30
31
Args:
32
status_code: HTTP status code (400-599)
33
detail: Error detail message or structured data
34
headers: Additional HTTP headers for error response
35
"""
36
self.status_code = status_code
37
self.detail = detail
38
self.headers = headers or {}
39
40
# Set exception message
41
if detail is None:
42
super().__init__(f"HTTP {status_code}")
43
else:
44
super().__init__(f"HTTP {status_code}: {detail}")
45
```
46
47
### ClientDisconnect Exception
48
49
```python { .api }
50
from starlette.requests import ClientDisconnect
51
52
class ClientDisconnect(Exception):
53
"""
54
Exception raised when client disconnects during request processing.
55
56
This can occur during streaming request body reads or when
57
the client closes the connection unexpectedly.
58
"""
59
pass
60
```
61
62
## WebSocket Exceptions
63
64
### WebSocket Exception Classes
65
66
```python { .api }
67
from starlette.websockets import WebSocketDisconnect
68
from starlette.exceptions import WebSocketException
69
70
class WebSocketDisconnect(Exception):
71
"""
72
Exception raised when WebSocket connection closes.
73
74
Contains close code and optional reason for disconnection.
75
"""
76
77
def __init__(self, code: int = 1000, reason: str | None = None) -> None:
78
"""
79
Initialize WebSocket disconnect exception.
80
81
Args:
82
code: WebSocket close code (1000 = normal closure)
83
reason: Optional close reason string
84
"""
85
self.code = code
86
self.reason = reason
87
88
super().__init__(f"WebSocket disconnect: {code}")
89
90
class WebSocketException(Exception):
91
"""
92
WebSocket error exception with close code and reason.
93
94
Used to signal WebSocket errors that should close the connection
95
with a specific code and reason.
96
"""
97
98
def __init__(self, code: int, reason: str | None = None) -> None:
99
"""
100
Initialize WebSocket exception.
101
102
Args:
103
code: WebSocket close code
104
reason: Optional close reason
105
"""
106
self.code = code
107
self.reason = reason
108
109
super().__init__(f"WebSocket error: {code}")
110
```
111
112
## HTTP Status Code Constants
113
114
### 1xx Informational Status Codes
115
116
```python { .api }
117
from starlette import status
118
119
# 1xx Informational responses
120
HTTP_100_CONTINUE = 100
121
HTTP_101_SWITCHING_PROTOCOLS = 101
122
HTTP_102_PROCESSING = 102
123
HTTP_103_EARLY_HINTS = 103
124
```
125
126
### 2xx Success Status Codes
127
128
```python { .api }
129
# 2xx Success responses
130
HTTP_200_OK = 200
131
HTTP_201_CREATED = 201
132
HTTP_202_ACCEPTED = 202
133
HTTP_203_NON_AUTHORITATIVE_INFORMATION = 203
134
HTTP_204_NO_CONTENT = 204
135
HTTP_205_RESET_CONTENT = 205
136
HTTP_206_PARTIAL_CONTENT = 206
137
HTTP_207_MULTI_STATUS = 207
138
HTTP_208_ALREADY_REPORTED = 208
139
HTTP_226_IM_USED = 226
140
```
141
142
### 3xx Redirection Status Codes
143
144
```python { .api }
145
# 3xx Redirection responses
146
HTTP_300_MULTIPLE_CHOICES = 300
147
HTTP_301_MOVED_PERMANENTLY = 301
148
HTTP_302_FOUND = 302
149
HTTP_303_SEE_OTHER = 303
150
HTTP_304_NOT_MODIFIED = 304
151
HTTP_305_USE_PROXY = 305
152
HTTP_306_RESERVED = 306
153
HTTP_307_TEMPORARY_REDIRECT = 307
154
HTTP_308_PERMANENT_REDIRECT = 308
155
```
156
157
### 4xx Client Error Status Codes
158
159
```python { .api }
160
# 4xx Client error responses
161
HTTP_400_BAD_REQUEST = 400
162
HTTP_401_UNAUTHORIZED = 401
163
HTTP_402_PAYMENT_REQUIRED = 402
164
HTTP_403_FORBIDDEN = 403
165
HTTP_404_NOT_FOUND = 404
166
HTTP_405_METHOD_NOT_ALLOWED = 405
167
HTTP_406_NOT_ACCEPTABLE = 406
168
HTTP_407_PROXY_AUTHENTICATION_REQUIRED = 407
169
HTTP_408_REQUEST_TIMEOUT = 408
170
HTTP_409_CONFLICT = 409
171
HTTP_410_GONE = 410
172
HTTP_411_LENGTH_REQUIRED = 411
173
HTTP_412_PRECONDITION_FAILED = 412
174
HTTP_413_REQUEST_ENTITY_TOO_LARGE = 413
175
HTTP_414_REQUEST_URI_TOO_LONG = 414
176
HTTP_415_UNSUPPORTED_MEDIA_TYPE = 415
177
HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE = 416
178
HTTP_417_EXPECTATION_FAILED = 417
179
HTTP_418_IM_A_TEAPOT = 418
180
HTTP_421_MISDIRECTED_REQUEST = 421
181
HTTP_422_UNPROCESSABLE_ENTITY = 422
182
HTTP_423_LOCKED = 423
183
HTTP_424_FAILED_DEPENDENCY = 424
184
HTTP_425_TOO_EARLY = 425
185
HTTP_426_UPGRADE_REQUIRED = 426
186
HTTP_428_PRECONDITION_REQUIRED = 428
187
HTTP_429_TOO_MANY_REQUESTS = 429
188
HTTP_431_REQUEST_HEADER_FIELDS_TOO_LARGE = 431
189
HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS = 451
190
```
191
192
### 5xx Server Error Status Codes
193
194
```python { .api }
195
# 5xx Server error responses
196
HTTP_500_INTERNAL_SERVER_ERROR = 500
197
HTTP_501_NOT_IMPLEMENTED = 501
198
HTTP_502_BAD_GATEWAY = 502
199
HTTP_503_SERVICE_UNAVAILABLE = 503
200
HTTP_504_GATEWAY_TIMEOUT = 504
201
HTTP_505_HTTP_VERSION_NOT_SUPPORTED = 505
202
HTTP_506_VARIANT_ALSO_NEGOTIATES = 506
203
HTTP_507_INSUFFICIENT_STORAGE = 507
204
HTTP_508_LOOP_DETECTED = 508
205
HTTP_510_NOT_EXTENDED = 510
206
HTTP_511_NETWORK_AUTHENTICATION_REQUIRED = 511
207
```
208
209
### WebSocket Status Code Constants
210
211
```python { .api }
212
# WebSocket close codes
213
WS_1000_NORMAL_CLOSURE = 1000
214
WS_1001_GOING_AWAY = 1001
215
WS_1002_PROTOCOL_ERROR = 1002
216
WS_1003_UNSUPPORTED_DATA = 1003
217
WS_1005_NO_STATUS_RCVD = 1005
218
WS_1006_ABNORMAL_CLOSURE = 1006
219
WS_1007_INVALID_FRAME_PAYLOAD_DATA = 1007
220
WS_1008_POLICY_VIOLATION = 1008
221
WS_1009_MESSAGE_TOO_BIG = 1009
222
WS_1010_MANDATORY_EXT = 1010
223
WS_1011_INTERNAL_ERROR = 1011
224
WS_1012_SERVICE_RESTART = 1012
225
WS_1013_TRY_AGAIN_LATER = 1013
226
WS_1014_BAD_GATEWAY = 1014
227
WS_1015_TLS_HANDSHAKE = 1015
228
```
229
230
## Exception Handling Patterns
231
232
### Basic HTTP Exception Usage
233
234
```python { .api }
235
from starlette.exceptions import HTTPException
236
from starlette import status
237
from starlette.responses import JSONResponse
238
239
async def get_user(request):
240
user_id = request.path_params["user_id"]
241
242
# Validate input
243
try:
244
user_id = int(user_id)
245
except ValueError:
246
raise HTTPException(
247
status_code=status.HTTP_400_BAD_REQUEST,
248
detail="Invalid user ID format"
249
)
250
251
# Check authorization
252
if not request.user.is_authenticated:
253
raise HTTPException(
254
status_code=status.HTTP_401_UNAUTHORIZED,
255
detail="Authentication required",
256
headers={"WWW-Authenticate": "Bearer"}
257
)
258
259
# Find user
260
user = await find_user_by_id(user_id)
261
if not user:
262
raise HTTPException(
263
status_code=status.HTTP_404_NOT_FOUND,
264
detail=f"User {user_id} not found"
265
)
266
267
# Check permissions
268
if user.id != request.user.id and not request.user.is_admin():
269
raise HTTPException(
270
status_code=status.HTTP_403_FORBIDDEN,
271
detail="Access denied"
272
)
273
274
return JSONResponse({"user": user.to_dict()})
275
```
276
277
### Structured Error Details
278
279
```python { .api }
280
async def create_user(request):
281
try:
282
data = await request.json()
283
except ValueError:
284
raise HTTPException(
285
status_code=status.HTTP_400_BAD_REQUEST,
286
detail="Invalid JSON in request body"
287
)
288
289
# Validation errors with structured details
290
errors = []
291
292
if not data.get("email"):
293
errors.append({"field": "email", "message": "Email is required"})
294
elif not is_valid_email(data["email"]):
295
errors.append({"field": "email", "message": "Invalid email format"})
296
297
if not data.get("username"):
298
errors.append({"field": "username", "message": "Username is required"})
299
elif len(data["username"]) < 3:
300
errors.append({"field": "username", "message": "Username too short"})
301
302
if errors:
303
raise HTTPException(
304
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
305
detail={
306
"message": "Validation failed",
307
"errors": errors
308
}
309
)
310
311
# Create user...
312
user = await create_user_in_db(data)
313
return JSONResponse(user.to_dict(), status_code=status.HTTP_201_CREATED)
314
```
315
316
### Custom Exception Classes
317
318
```python { .api }
319
class ValidationError(HTTPException):
320
"""Validation error with field-specific details."""
321
322
def __init__(self, errors: list[dict]):
323
super().__init__(
324
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
325
detail={
326
"message": "Validation failed",
327
"errors": errors
328
}
329
)
330
331
class NotFoundError(HTTPException):
332
"""Resource not found error."""
333
334
def __init__(self, resource: str, identifier: str | int):
335
super().__init__(
336
status_code=status.HTTP_404_NOT_FOUND,
337
detail=f"{resource} '{identifier}' not found"
338
)
339
340
class PermissionDeniedError(HTTPException):
341
"""Permission denied error."""
342
343
def __init__(self, action: str, resource: str = "resource"):
344
super().__init__(
345
status_code=status.HTTP_403_FORBIDDEN,
346
detail=f"Permission denied: cannot {action} {resource}"
347
)
348
349
class RateLimitError(HTTPException):
350
"""Rate limit exceeded error."""
351
352
def __init__(self, retry_after: int = 60):
353
super().__init__(
354
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
355
detail="Rate limit exceeded",
356
headers={"Retry-After": str(retry_after)}
357
)
358
359
# Usage
360
async def delete_user(request):
361
user_id = request.path_params["user_id"]
362
363
user = await find_user_by_id(user_id)
364
if not user:
365
raise NotFoundError("User", user_id)
366
367
if not can_delete_user(request.user, user):
368
raise PermissionDeniedError("delete", "user")
369
370
await delete_user_from_db(user_id)
371
return JSONResponse({"message": "User deleted"})
372
```
373
374
## WebSocket Exception Handling
375
376
### WebSocket Error Handling
377
378
```python { .api }
379
from starlette.websockets import WebSocket, WebSocketDisconnect
380
from starlette.exceptions import WebSocketException
381
from starlette import status as ws_status
382
383
async def websocket_endpoint(websocket: WebSocket):
384
try:
385
# Authentication check
386
token = websocket.query_params.get("token")
387
if not token:
388
raise WebSocketException(
389
code=ws_status.WS_1008_POLICY_VIOLATION,
390
reason="Authentication token required"
391
)
392
393
user = await authenticate_token(token)
394
if not user:
395
raise WebSocketException(
396
code=ws_status.WS_1008_POLICY_VIOLATION,
397
reason="Invalid authentication token"
398
)
399
400
await websocket.accept()
401
402
# Handle messages
403
while True:
404
try:
405
data = await websocket.receive_json()
406
407
# Validate message
408
if not isinstance(data, dict):
409
await websocket.send_json({
410
"type": "error",
411
"message": "Message must be JSON object"
412
})
413
continue
414
415
# Process message
416
await process_websocket_message(websocket, user, data)
417
418
except ValueError:
419
# Invalid JSON
420
await websocket.send_json({
421
"type": "error",
422
"message": "Invalid JSON format"
423
})
424
425
except WebSocketDisconnect:
426
print(f"Client disconnected")
427
428
except WebSocketException as e:
429
# Close with specific code and reason
430
await websocket.close(code=e.code, reason=e.reason)
431
432
except Exception as e:
433
# Unexpected error - close with internal error code
434
print(f"Unexpected error: {e}")
435
await websocket.close(
436
code=ws_status.WS_1011_INTERNAL_ERROR,
437
reason="Internal server error"
438
)
439
440
async def process_websocket_message(websocket: WebSocket, user, data):
441
message_type = data.get("type")
442
443
if message_type == "ping":
444
await websocket.send_json({"type": "pong"})
445
446
elif message_type == "chat_message":
447
if not data.get("content"):
448
await websocket.send_json({
449
"type": "error",
450
"message": "Message content required"
451
})
452
return
453
454
# Process chat message...
455
await broadcast_chat_message(user, data["content"])
456
457
else:
458
await websocket.send_json({
459
"type": "error",
460
"message": f"Unknown message type: {message_type}"
461
})
462
```
463
464
## Global Exception Handling
465
466
### Application-level Exception Handlers
467
468
```python { .api }
469
from starlette.applications import Starlette
470
from starlette.middleware.exceptions import ExceptionMiddleware
471
from starlette.responses import JSONResponse, PlainTextResponse
472
from starlette.requests import Request
473
474
# Custom exception handlers
475
async def http_exception_handler(request: Request, exc: HTTPException):
476
"""Handle HTTPException instances."""
477
return JSONResponse(
478
content={
479
"error": {
480
"status_code": exc.status_code,
481
"detail": exc.detail,
482
"type": "HTTPException"
483
}
484
},
485
status_code=exc.status_code,
486
headers=exc.headers
487
)
488
489
async def validation_error_handler(request: Request, exc: ValidationError):
490
"""Handle custom validation errors."""
491
return JSONResponse(
492
content={
493
"error": {
494
"type": "ValidationError",
495
"message": "Request validation failed",
496
"details": exc.detail
497
}
498
},
499
status_code=exc.status_code
500
)
501
502
async def generic_error_handler(request: Request, exc: Exception):
503
"""Handle unexpected exceptions."""
504
print(f"Unexpected error: {exc}")
505
506
return JSONResponse(
507
content={
508
"error": {
509
"type": "InternalServerError",
510
"message": "An unexpected error occurred"
511
}
512
},
513
status_code=500
514
)
515
516
async def not_found_handler(request: Request, exc):
517
"""Handle 404 errors."""
518
return JSONResponse(
519
content={
520
"error": {
521
"type": "NotFound",
522
"message": "The requested resource was not found",
523
"path": request.url.path
524
}
525
},
526
status_code=404
527
)
528
529
# Configure exception handlers
530
exception_handlers = {
531
HTTPException: http_exception_handler,
532
ValidationError: validation_error_handler,
533
404: not_found_handler, # By status code
534
500: generic_error_handler, # By status code
535
}
536
537
app = Starlette(
538
routes=routes,
539
exception_handlers=exception_handlers
540
)
541
542
# Add exception handlers dynamically
543
app.add_exception_handler(ValueError, validation_error_handler)
544
app.add_exception_handler(KeyError, generic_error_handler)
545
```
546
547
### Development vs Production Error Handling
548
549
```python { .api }
550
from starlette.config import Config
551
from starlette.responses import HTMLResponse
552
import traceback
553
554
config = Config()
555
DEBUG = config("DEBUG", cast=bool, default=False)
556
557
async def debug_exception_handler(request: Request, exc: Exception):
558
"""Detailed error handler for development."""
559
error_html = f"""
560
<html>
561
<head><title>Error: {exc.__class__.__name__}</title></head>
562
<body>
563
<h1>Error: {exc.__class__.__name__}</h1>
564
<p><strong>Message:</strong> {str(exc)}</p>
565
<h2>Traceback:</h2>
566
<pre>{traceback.format_exc()}</pre>
567
<h2>Request Info:</h2>
568
<pre>
569
Method: {request.method}
570
URL: {request.url}
571
Headers: {dict(request.headers)}
572
</pre>
573
</body>
574
</html>
575
"""
576
return HTMLResponse(error_html, status_code=500)
577
578
async def production_exception_handler(request: Request, exc: Exception):
579
"""Safe error handler for production."""
580
# Log the error (use proper logging in production)
581
print(f"ERROR: {exc}")
582
583
return JSONResponse(
584
content={
585
"error": {
586
"type": "InternalServerError",
587
"message": "An internal error occurred"
588
}
589
},
590
status_code=500
591
)
592
593
# Choose handler based on environment
594
exception_handler = debug_exception_handler if DEBUG else production_exception_handler
595
596
app.add_exception_handler(Exception, exception_handler)
597
```
598
599
## Exception Middleware Integration
600
601
### Custom Exception Middleware
602
603
```python { .api }
604
from starlette.middleware.base import BaseHTTPMiddleware
605
import logging
606
607
class ErrorLoggingMiddleware(BaseHTTPMiddleware):
608
"""Middleware to log exceptions with context."""
609
610
async def dispatch(self, request, call_next):
611
try:
612
response = await call_next(request)
613
return response
614
615
except HTTPException as exc:
616
# Log HTTP exceptions with request context
617
logging.warning(
618
f"HTTP {exc.status_code}: {exc.detail}",
619
extra={
620
"method": request.method,
621
"path": request.url.path,
622
"client": request.client.host if request.client else None,
623
"user_agent": request.headers.get("user-agent"),
624
}
625
)
626
raise # Re-raise to be handled by exception handlers
627
628
except Exception as exc:
629
# Log unexpected exceptions
630
logging.error(
631
f"Unexpected error: {exc}",
632
exc_info=True,
633
extra={
634
"method": request.method,
635
"path": request.url.path,
636
"client": request.client.host if request.client else None,
637
}
638
)
639
raise
640
641
# Add to middleware stack
642
app.add_middleware(ErrorLoggingMiddleware)
643
```
644
645
## Testing Exception Handling
646
647
### Testing HTTP Exceptions
648
649
```python { .api }
650
from starlette.testclient import TestClient
651
import pytest
652
653
def test_http_exceptions():
654
client = TestClient(app)
655
656
# Test 404 Not Found
657
response = client.get("/nonexistent")
658
assert response.status_code == 404
659
assert "error" in response.json()
660
661
# Test 400 Bad Request
662
response = client.post("/users", json={"invalid": "data"})
663
assert response.status_code == 400
664
665
# Test 401 Unauthorized
666
response = client.get("/protected")
667
assert response.status_code == 401
668
669
# Test 403 Forbidden
670
response = client.get("/admin", headers={"Authorization": "Bearer user-token"})
671
assert response.status_code == 403
672
673
def test_custom_exceptions():
674
client = TestClient(app)
675
676
# Test validation error
677
response = client.post("/users", json={})
678
assert response.status_code == 422
679
680
error_data = response.json()["error"]
681
assert error_data["type"] == "ValidationError"
682
assert "errors" in error_data["details"]
683
684
def test_exception_handlers():
685
client = TestClient(app)
686
687
# Test that exceptions are properly handled
688
response = client.get("/trigger-error")
689
assert response.status_code == 500
690
691
# Verify error format
692
error = response.json()["error"]
693
assert error["type"] == "InternalServerError"
694
```
695
696
### Testing WebSocket Exceptions
697
698
```python { .api }
699
def test_websocket_exceptions():
700
client = TestClient(app)
701
702
# Test authentication failure
703
with pytest.raises(WebSocketDenialResponse):
704
with client.websocket_connect("/ws"):
705
pass
706
707
# Test successful connection
708
with client.websocket_connect("/ws?token=valid") as websocket:
709
websocket.send_json({"type": "ping"})
710
response = websocket.receive_json()
711
assert response["type"] == "pong"
712
713
# Test invalid message handling
714
websocket.send_text("invalid json")
715
error_response = websocket.receive_json()
716
assert error_response["type"] == "error"
717
```
718
719
Starlette's exception handling system provides comprehensive error management with flexible handlers, structured error responses, and proper integration with HTTP and WebSocket protocols for building robust, user-friendly applications.