0
# Exception Handling
1
2
FastAPI provides comprehensive exception handling with automatic HTTP response generation, custom exception classes, and flexible error handling patterns. Exceptions are automatically converted to appropriate HTTP responses with proper status codes and error details.
3
4
## Capabilities
5
6
### HTTP Exception
7
8
Primary exception class for HTTP errors with automatic response generation and OpenAPI documentation.
9
10
```python { .api }
11
class HTTPException(Exception):
12
def __init__(
13
self,
14
status_code: int,
15
detail: Any = None,
16
headers: Optional[Dict[str, str]] = None
17
) -> None:
18
"""
19
HTTP exception for client and server errors.
20
21
Parameters:
22
- status_code: HTTP status code (400-599)
23
- detail: Error detail message or structured data
24
Can be string, dict, list, or any JSON-serializable data
25
- headers: Additional HTTP headers to include in error response
26
27
Behaviors:
28
- Automatically generates JSON error response
29
- Includes status code and detail in response body
30
- Supports structured error details with validation info
31
- Integrates with OpenAPI documentation for error responses
32
- Can be caught and handled by custom exception handlers
33
"""
34
```
35
36
### WebSocket Exception
37
38
Exception class for WebSocket connection errors with proper close codes.
39
40
```python { .api }
41
class WebSocketException(Exception):
42
def __init__(self, code: int, reason: Optional[str] = None) -> None:
43
"""
44
WebSocket connection exception.
45
46
Parameters:
47
- code: WebSocket close code (1000-4999)
48
Standard codes: 1000 (normal), 1001 (going away), 1002 (protocol error), etc.
49
- reason: Optional reason string for the close
50
51
Behaviors:
52
- Automatically closes WebSocket connection with specified code
53
- Sends close frame with code and reason to client
54
- Follows WebSocket RFC close code standards
55
- Can be caught for custom WebSocket error handling
56
"""
57
```
58
59
### Validation Exception Classes
60
61
Exception classes for request and response validation errors with detailed field-level error information.
62
63
```python { .api }
64
class RequestValidationError(ValueError):
65
"""
66
Exception raised when request data validation fails.
67
68
Contains detailed information about validation errors including:
69
- Field names and locations (path, query, header, body)
70
- Validation error types and messages
71
- Input values that caused errors
72
73
Automatically generates 422 Unprocessable Entity responses
74
with structured error details.
75
"""
76
77
class ResponseValidationError(ValueError):
78
"""
79
Exception raised when response data validation fails.
80
81
Indicates programming errors where endpoint returns data
82
that doesn't match the declared response model.
83
84
Automatically generates 500 Internal Server Error responses
85
in development, helps catch response model mismatches.
86
"""
87
88
class WebSocketRequestValidationError(ValueError):
89
"""
90
Exception raised when WebSocket request validation fails.
91
92
Similar to RequestValidationError but for WebSocket connections.
93
Automatically closes WebSocket connection with error details.
94
"""
95
```
96
97
### Base FastAPI Exception
98
99
Base exception class for FastAPI-specific errors.
100
101
```python { .api }
102
class FastAPIError(Exception):
103
"""
104
Base exception class for FastAPI framework errors.
105
106
Used for framework-level errors and as base class
107
for other FastAPI-specific exceptions.
108
"""
109
```
110
111
## Exception Handler System
112
113
### Custom Exception Handlers
114
115
Functions for handling specific exceptions with custom logic and responses.
116
117
```python { .api }
118
@app.exception_handler(ExceptionClass)
119
async def custom_exception_handler(request: Request, exc: ExceptionClass) -> Response:
120
"""
121
Custom exception handler function.
122
123
Parameters:
124
- request: The HTTP request that caused the exception
125
- exc: The exception instance that was raised
126
127
Returns:
128
Custom Response object with appropriate status code and content
129
130
Behaviors:
131
- Called automatically when specified exception is raised
132
- Can return any Response type (JSON, HTML, plain text, etc.)
133
- Has access to full request context for logging or custom logic
134
- Can modify response headers and status codes
135
- Supports both sync and async handler functions
136
"""
137
```
138
139
## Usage Examples
140
141
### Basic HTTP Exceptions
142
143
```python
144
from fastapi import FastAPI, HTTPException, status
145
146
app = FastAPI()
147
148
@app.get("/items/{item_id}")
149
async def get_item(item_id: int):
150
if item_id < 1:
151
raise HTTPException(
152
status_code=400,
153
detail="Item ID must be positive"
154
)
155
156
if item_id > 1000:
157
raise HTTPException(
158
status_code=404,
159
detail="Item not found"
160
)
161
162
# Simulate item not found
163
if item_id == 999:
164
raise HTTPException(
165
status_code=404,
166
detail={
167
"error": "Item not found",
168
"item_id": item_id,
169
"suggestion": "Try a different item ID"
170
}
171
)
172
173
return {"item_id": item_id, "name": f"Item {item_id}"}
174
175
@app.post("/users/")
176
async def create_user(user_data: dict):
177
# Check for duplicate username
178
if user_data.get("username") == "admin":
179
raise HTTPException(
180
status_code=409,
181
detail="Username 'admin' is reserved",
182
headers={"X-Error": "username-conflict"}
183
)
184
185
# Validate user data
186
if not user_data.get("email"):
187
raise HTTPException(
188
status_code=422,
189
detail={
190
"field": "email",
191
"message": "Email is required",
192
"code": "missing_field"
193
}
194
)
195
196
return {"message": "User created successfully"}
197
```
198
199
### Custom Exception Classes
200
201
```python
202
from fastapi import FastAPI, Request, HTTPException
203
from fastapi.responses import JSONResponse
204
205
app = FastAPI()
206
207
class UserNotFoundError(Exception):
208
def __init__(self, user_id: int):
209
self.user_id = user_id
210
self.message = f"User {user_id} not found"
211
super().__init__(self.message)
212
213
class InsufficientPermissionsError(Exception):
214
def __init__(self, required_role: str, user_role: str):
215
self.required_role = required_role
216
self.user_role = user_role
217
self.message = f"Required role: {required_role}, user role: {user_role}"
218
super().__init__(self.message)
219
220
class DatabaseConnectionError(Exception):
221
def __init__(self, database: str):
222
self.database = database
223
self.message = f"Failed to connect to database: {database}"
224
super().__init__(self.message)
225
226
# Custom exception handlers
227
@app.exception_handler(UserNotFoundError)
228
async def user_not_found_handler(request: Request, exc: UserNotFoundError):
229
return JSONResponse(
230
status_code=404,
231
content={
232
"error": "user_not_found",
233
"message": exc.message,
234
"user_id": exc.user_id,
235
"timestamp": "2023-01-01T00:00:00Z"
236
}
237
)
238
239
@app.exception_handler(InsufficientPermissionsError)
240
async def insufficient_permissions_handler(request: Request, exc: InsufficientPermissionsError):
241
return JSONResponse(
242
status_code=403,
243
content={
244
"error": "insufficient_permissions",
245
"message": exc.message,
246
"required_role": exc.required_role,
247
"user_role": exc.user_role
248
}
249
)
250
251
@app.exception_handler(DatabaseConnectionError)
252
async def database_error_handler(request: Request, exc: DatabaseConnectionError):
253
# Log the error
254
print(f"Database connection failed: {exc.database}")
255
256
return JSONResponse(
257
status_code=503,
258
content={
259
"error": "service_unavailable",
260
"message": "Database temporarily unavailable",
261
"retry_after": 60
262
},
263
headers={"Retry-After": "60"}
264
)
265
266
# Using custom exceptions
267
@app.get("/users/{user_id}")
268
async def get_user(user_id: int):
269
# Simulate database connection check
270
if not check_database_connection():
271
raise DatabaseConnectionError("user_db")
272
273
# Simulate user lookup
274
user = find_user(user_id) # Your database query
275
if not user:
276
raise UserNotFoundError(user_id)
277
278
return user
279
280
@app.delete("/users/{user_id}")
281
async def delete_user(user_id: int, current_user: dict):
282
# Check permissions
283
if current_user.get("role") != "admin":
284
raise InsufficientPermissionsError("admin", current_user.get("role", "user"))
285
286
# Check if user exists
287
if not user_exists(user_id):
288
raise UserNotFoundError(user_id)
289
290
# Delete user
291
delete_user_from_db(user_id)
292
return {"message": "User deleted successfully"}
293
```
294
295
### WebSocket Exception Handling
296
297
```python
298
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, WebSocketException
299
import json
300
301
app = FastAPI()
302
303
@app.websocket("/ws/{client_id}")
304
async def websocket_endpoint(websocket: WebSocket, client_id: str):
305
await websocket.accept()
306
307
try:
308
while True:
309
# Receive message
310
data = await websocket.receive_text()
311
312
try:
313
# Parse JSON message
314
message = json.loads(data)
315
except json.JSONDecodeError:
316
# Send error and close connection
317
raise WebSocketException(
318
code=1003, # Unsupported data
319
reason="Invalid JSON format"
320
)
321
322
# Validate message structure
323
if "type" not in message:
324
raise WebSocketException(
325
code=1002, # Protocol error
326
reason="Message must include 'type' field"
327
)
328
329
# Handle different message types
330
if message["type"] == "ping":
331
await websocket.send_text(json.dumps({"type": "pong"}))
332
333
elif message["type"] == "auth":
334
# Validate authentication
335
if not validate_token(message.get("token")):
336
raise WebSocketException(
337
code=1008, # Policy violation
338
reason="Invalid authentication token"
339
)
340
await websocket.send_text(json.dumps({"type": "auth_success"}))
341
342
elif message["type"] == "data":
343
# Process data message
344
result = process_data(message.get("payload"))
345
await websocket.send_text(json.dumps({
346
"type": "result",
347
"data": result
348
}))
349
350
else:
351
raise WebSocketException(
352
code=1003, # Unsupported data
353
reason=f"Unknown message type: {message['type']}"
354
)
355
356
except WebSocketDisconnect:
357
print(f"Client {client_id} disconnected normally")
358
359
except WebSocketException:
360
print(f"Client {client_id} disconnected due to protocol error")
361
# Exception automatically closes connection with proper code
362
363
except Exception as e:
364
print(f"Unexpected error for client {client_id}: {e}")
365
# Close connection with internal error code
366
await websocket.close(code=1011, reason="Internal server error")
367
```
368
369
### Global Exception Handlers
370
371
```python
372
from fastapi import FastAPI, Request, HTTPException
373
from fastapi.responses import JSONResponse
374
from fastapi.exceptions import RequestValidationError, ResponseValidationError
375
import logging
376
377
app = FastAPI()
378
379
# Set up logging
380
logging.basicConfig(level=logging.INFO)
381
logger = logging.getLogger(__name__)
382
383
# Global exception handler for unhandled exceptions
384
@app.exception_handler(Exception)
385
async def global_exception_handler(request: Request, exc: Exception):
386
logger.error(f"Unhandled exception: {exc}", exc_info=True)
387
388
return JSONResponse(
389
status_code=500,
390
content={
391
"error": "internal_server_error",
392
"message": "An unexpected error occurred",
393
"request_id": generate_request_id() # Your ID generation logic
394
}
395
)
396
397
# Custom handler for validation errors
398
@app.exception_handler(RequestValidationError)
399
async def validation_exception_handler(request: Request, exc: RequestValidationError):
400
# Log validation errors
401
logger.warning(f"Validation error on {request.url}: {exc.errors()}")
402
403
# Transform validation errors into custom format
404
errors = []
405
for error in exc.errors():
406
errors.append({
407
"field": ".".join(str(x) for x in error["loc"]),
408
"message": error["msg"],
409
"type": error["type"],
410
"input": error.get("input")
411
})
412
413
return JSONResponse(
414
status_code=422,
415
content={
416
"error": "validation_failed",
417
"message": "Request validation failed",
418
"details": errors,
419
"request_id": generate_request_id()
420
}
421
)
422
423
# Handler for response validation errors (development only)
424
@app.exception_handler(ResponseValidationError)
425
async def response_validation_exception_handler(request: Request, exc: ResponseValidationError):
426
logger.error(f"Response validation error: {exc}", exc_info=True)
427
428
return JSONResponse(
429
status_code=500,
430
content={
431
"error": "response_validation_failed",
432
"message": "Response data doesn't match declared model",
433
"detail": str(exc) if app.debug else "Internal server error"
434
}
435
)
436
437
# Override default HTTP exception handler
438
@app.exception_handler(HTTPException)
439
async def http_exception_handler(request: Request, exc: HTTPException):
440
# Log HTTP exceptions
441
logger.info(f"HTTP {exc.status_code} on {request.url}: {exc.detail}")
442
443
# Add request ID to all HTTP exceptions
444
content = {
445
"error": "http_exception",
446
"message": exc.detail,
447
"status_code": exc.status_code,
448
"request_id": generate_request_id()
449
}
450
451
headers = exc.headers or {}
452
headers["X-Request-ID"] = content["request_id"]
453
454
return JSONResponse(
455
status_code=exc.status_code,
456
content=content,
457
headers=headers
458
)
459
```
460
461
### Conditional Exception Handling
462
463
```python
464
from fastapi import FastAPI, Request, HTTPException
465
from fastapi.responses import JSONResponse, HTMLResponse
466
import os
467
468
app = FastAPI()
469
470
# Environment-specific exception handling
471
IS_DEVELOPMENT = os.getenv("ENVIRONMENT") == "development"
472
473
@app.exception_handler(Exception)
474
async def environment_aware_handler(request: Request, exc: Exception):
475
# Different handling based on environment
476
if IS_DEVELOPMENT:
477
# Detailed error information in development
478
import traceback
479
return JSONResponse(
480
status_code=500,
481
content={
482
"error": "internal_server_error",
483
"message": str(exc),
484
"type": type(exc).__name__,
485
"traceback": traceback.format_exc().split("\n")
486
}
487
)
488
else:
489
# Minimal error information in production
490
logger.error(f"Production error: {exc}", exc_info=True)
491
return JSONResponse(
492
status_code=500,
493
content={
494
"error": "internal_server_error",
495
"message": "An unexpected error occurred"
496
}
497
)
498
499
# Content-type specific error responses
500
@app.exception_handler(HTTPException)
501
async def content_type_aware_handler(request: Request, exc: HTTPException):
502
accept_header = request.headers.get("accept", "")
503
504
if "text/html" in accept_header:
505
# Return HTML error page for browsers
506
html_content = f"""
507
<html>
508
<head><title>Error {exc.status_code}</title></head>
509
<body>
510
<h1>Error {exc.status_code}</h1>
511
<p>{exc.detail}</p>
512
<a href="/">Go back to home</a>
513
</body>
514
</html>
515
"""
516
return HTMLResponse(content=html_content, status_code=exc.status_code)
517
518
else:
519
# Return JSON for API clients
520
return JSONResponse(
521
status_code=exc.status_code,
522
content={
523
"error": f"http_{exc.status_code}",
524
"message": exc.detail
525
}
526
)
527
```
528
529
### Error Logging and Monitoring
530
531
```python
532
from fastapi import FastAPI, Request, HTTPException
533
from fastapi.responses import JSONResponse
534
import logging
535
import time
536
import uuid
537
538
app = FastAPI()
539
540
# Set up structured logging
541
logging.basicConfig(
542
level=logging.INFO,
543
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
544
)
545
logger = logging.getLogger(__name__)
546
547
def generate_request_id():
548
return str(uuid.uuid4())
549
550
@app.middleware("http")
551
async def add_request_id(request: Request, call_next):
552
request_id = generate_request_id()
553
request.state.request_id = request_id
554
555
start_time = time.time()
556
557
response = await call_next(request)
558
559
process_time = time.time() - start_time
560
561
# Log request completion
562
logger.info(
563
f"Request completed",
564
extra={
565
"request_id": request_id,
566
"method": request.method,
567
"url": str(request.url),
568
"status_code": response.status_code,
569
"process_time": process_time
570
}
571
)
572
573
response.headers["X-Request-ID"] = request_id
574
return response
575
576
@app.exception_handler(HTTPException)
577
async def logged_http_exception_handler(request: Request, exc: HTTPException):
578
request_id = getattr(request.state, "request_id", "unknown")
579
580
# Log HTTP exceptions with context
581
logger.warning(
582
f"HTTP exception occurred",
583
extra={
584
"request_id": request_id,
585
"method": request.method,
586
"url": str(request.url),
587
"status_code": exc.status_code,
588
"detail": exc.detail,
589
"user_agent": request.headers.get("user-agent"),
590
"client_ip": request.client.host if request.client else None
591
}
592
)
593
594
return JSONResponse(
595
status_code=exc.status_code,
596
content={
597
"error": f"http_{exc.status_code}",
598
"message": exc.detail,
599
"request_id": request_id
600
},
601
headers={"X-Request-ID": request_id}
602
)
603
604
@app.exception_handler(Exception)
605
async def logged_exception_handler(request: Request, exc: Exception):
606
request_id = getattr(request.state, "request_id", "unknown")
607
608
# Log unexpected exceptions with full context
609
logger.error(
610
f"Unexpected exception occurred",
611
extra={
612
"request_id": request_id,
613
"method": request.method,
614
"url": str(request.url),
615
"exception_type": type(exc).__name__,
616
"exception_message": str(exc),
617
"user_agent": request.headers.get("user-agent"),
618
"client_ip": request.client.host if request.client else None
619
},
620
exc_info=True
621
)
622
623
return JSONResponse(
624
status_code=500,
625
content={
626
"error": "internal_server_error",
627
"message": "An unexpected error occurred",
628
"request_id": request_id
629
},
630
headers={"X-Request-ID": request_id}
631
)
632
```