0
# Exception Handling
1
2
FastAPI provides comprehensive exception handling capabilities that integrate with HTTP status codes, automatic error responses, and custom exception handlers. The framework includes built-in exceptions for common scenarios and allows for custom exception handling patterns.
3
4
## Capabilities
5
6
### HTTP Exception
7
8
The primary exception class for handling HTTP errors with proper status codes and response formatting.
9
10
```python { .api }
11
class HTTPException(Exception):
12
def __init__(
13
self,
14
status_code: int,
15
detail: Any = None,
16
headers: Dict[str, str] = None
17
) -> None:
18
"""
19
HTTP exception with status code and detail message.
20
21
Parameters:
22
- status_code: HTTP status code (400, 401, 404, 500, etc.)
23
- detail: Error detail message or structured data
24
- headers: Additional HTTP headers to include in error response
25
"""
26
self.status_code = status_code
27
self.detail = detail
28
self.headers = headers
29
```
30
31
### WebSocket Exception
32
33
Exception class specifically for WebSocket connection errors.
34
35
```python { .api }
36
class WebSocketException(Exception):
37
def __init__(
38
self,
39
code: int,
40
reason: str = None
41
) -> None:
42
"""
43
WebSocket error exception.
44
45
Parameters:
46
- code: WebSocket close code (1000-4999)
47
- reason: Human-readable close reason
48
"""
49
self.code = code
50
self.reason = reason
51
```
52
53
### FastAPI Error
54
55
Generic runtime error exception for FastAPI framework issues.
56
57
```python { .api }
58
class FastAPIError(Exception):
59
def __init__(self, message: str = None) -> None:
60
"""
61
Generic FastAPI runtime error.
62
63
Parameters:
64
- message: Error message describing the issue
65
"""
66
self.message = message
67
```
68
69
### Validation Exceptions
70
71
Exception classes for handling request and response validation errors with detailed error information.
72
73
```python { .api }
74
class ValidationException(Exception):
75
def __init__(self, errors: List[dict]) -> None:
76
"""
77
Base validation error exception.
78
79
Parameters:
80
- errors: List of validation error dictionaries
81
"""
82
self.errors = errors
83
84
class RequestValidationError(ValidationException):
85
def __init__(self, errors: List[ErrorWrapper]) -> None:
86
"""
87
Request validation error with detailed error information.
88
89
Parameters:
90
- errors: List of Pydantic ErrorWrapper objects containing
91
field-specific validation error details
92
"""
93
self.errors = errors
94
self.body = getattr(errors, "body", None)
95
96
class WebSocketRequestValidationError(ValidationException):
97
def __init__(self, errors: List[ErrorWrapper]) -> None:
98
"""
99
WebSocket request validation error.
100
101
Parameters:
102
- errors: List of Pydantic ErrorWrapper objects for WebSocket
103
parameter validation failures
104
"""
105
self.errors = errors
106
107
class ResponseValidationError(ValidationException):
108
def __init__(self, errors: List[ErrorWrapper]) -> None:
109
"""
110
Response validation error for response model validation failures.
111
112
Parameters:
113
- errors: List of Pydantic ErrorWrapper objects containing
114
response validation error details
115
"""
116
self.errors = errors
117
```
118
119
### Exception Handler Decorator
120
121
Decorator method for registering custom exception handlers on FastAPI and APIRouter instances.
122
123
```python { .api }
124
def exception_handler(
125
self,
126
exc_class_or_status_code: Union[int, Type[Exception]]
127
) -> Callable[[Callable], Callable]:
128
"""
129
Decorator for adding exception handlers.
130
131
Parameters:
132
- exc_class_or_status_code: Exception class or HTTP status code to handle
133
134
Returns:
135
Decorator function for exception handler registration
136
"""
137
```
138
139
### Exception Handler Function Type
140
141
Type signature for custom exception handler functions.
142
143
```python { .api }
144
async def exception_handler_function(
145
request: Request,
146
exc: Exception
147
) -> Response:
148
"""
149
Exception handler function signature.
150
151
Parameters:
152
- request: HTTP request object
153
- exc: Exception instance that was raised
154
155
Returns:
156
Response object to send to client
157
"""
158
```
159
160
## Usage Examples
161
162
### Basic HTTP Exception Handling
163
164
```python
165
from fastapi import FastAPI, HTTPException, status
166
167
app = FastAPI()
168
169
@app.get("/items/{item_id}")
170
async def read_item(item_id: int):
171
if item_id == 0:
172
raise HTTPException(
173
status_code=status.HTTP_404_NOT_FOUND,
174
detail="Item not found"
175
)
176
if item_id < 0:
177
raise HTTPException(
178
status_code=status.HTTP_400_BAD_REQUEST,
179
detail="Item ID must be positive",
180
headers={"X-Error": "Invalid item ID"}
181
)
182
return {"item_id": item_id, "name": f"Item {item_id}"}
183
```
184
185
### Custom Exception Handler
186
187
```python
188
from fastapi import FastAPI, Request, HTTPException
189
from fastapi.responses import JSONResponse
190
from fastapi.exceptions import RequestValidationError
191
from starlette.exceptions import HTTPException as StarletteHTTPException
192
193
app = FastAPI()
194
195
@app.exception_handler(StarletteHTTPException)
196
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
197
return JSONResponse(
198
status_code=exc.status_code,
199
content={
200
"error": "HTTP Exception",
201
"message": exc.detail,
202
"path": request.url.path,
203
"method": request.method
204
}
205
)
206
207
@app.exception_handler(RequestValidationError)
208
async def validation_exception_handler(request: Request, exc: RequestValidationError):
209
return JSONResponse(
210
status_code=422,
211
content={
212
"error": "Validation Error",
213
"message": "Invalid request data",
214
"details": exc.errors(),
215
"body": exc.body
216
}
217
)
218
219
@app.get("/items/{item_id}")
220
async def read_item(item_id: int):
221
if item_id == 42:
222
raise HTTPException(status_code=418, detail="I'm a teapot")
223
return {"item_id": item_id}
224
```
225
226
### Custom Exception Classes
227
228
```python
229
from fastapi import FastAPI, Request, HTTPException
230
from fastapi.responses import JSONResponse
231
232
app = FastAPI()
233
234
# Custom exception classes
235
class ItemNotFoundError(Exception):
236
def __init__(self, item_id: int):
237
self.item_id = item_id
238
self.message = f"Item with ID {item_id} not found"
239
super().__init__(self.message)
240
241
class InsufficientPermissionsError(Exception):
242
def __init__(self, required_role: str, user_role: str):
243
self.required_role = required_role
244
self.user_role = user_role
245
self.message = f"Required role: {required_role}, user role: {user_role}"
246
super().__init__(self.message)
247
248
class BusinessLogicError(Exception):
249
def __init__(self, message: str, error_code: str):
250
self.message = message
251
self.error_code = error_code
252
super().__init__(self.message)
253
254
# Exception handlers
255
@app.exception_handler(ItemNotFoundError)
256
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
257
return JSONResponse(
258
status_code=404,
259
content={
260
"error": "Item Not Found",
261
"message": exc.message,
262
"item_id": exc.item_id
263
}
264
)
265
266
@app.exception_handler(InsufficientPermissionsError)
267
async def insufficient_permissions_handler(request: Request, exc: InsufficientPermissionsError):
268
return JSONResponse(
269
status_code=403,
270
content={
271
"error": "Insufficient Permissions",
272
"message": exc.message,
273
"required_role": exc.required_role,
274
"user_role": exc.user_role
275
}
276
)
277
278
@app.exception_handler(BusinessLogicError)
279
async def business_logic_handler(request: Request, exc: BusinessLogicError):
280
return JSONResponse(
281
status_code=422,
282
content={
283
"error": "Business Logic Error",
284
"message": exc.message,
285
"error_code": exc.error_code
286
}
287
)
288
289
# Routes using custom exceptions
290
@app.get("/items/{item_id}")
291
async def get_item(item_id: int):
292
if item_id == 999:
293
raise ItemNotFoundError(item_id)
294
return {"item_id": item_id, "name": f"Item {item_id}"}
295
296
@app.delete("/items/{item_id}")
297
async def delete_item(item_id: int, user_role: str = "user"):
298
if user_role != "admin":
299
raise InsufficientPermissionsError("admin", user_role)
300
if item_id <= 0:
301
raise BusinessLogicError("Cannot delete items with non-positive IDs", "INVALID_ID")
302
return {"message": f"Item {item_id} deleted"}
303
```
304
305
### Exception Handling with Logging
306
307
```python
308
import logging
309
from datetime import datetime
310
from fastapi import FastAPI, Request, HTTPException
311
from fastapi.responses import JSONResponse
312
313
# Configure logging
314
logging.basicConfig(level=logging.INFO)
315
logger = logging.getLogger(__name__)
316
317
app = FastAPI()
318
319
@app.exception_handler(Exception)
320
async def global_exception_handler(request: Request, exc: Exception):
321
# Log the exception
322
logger.error(
323
f"Unhandled exception: {type(exc).__name__}: {str(exc)}",
324
extra={
325
"path": request.url.path,
326
"method": request.method,
327
"timestamp": datetime.utcnow().isoformat()
328
}
329
)
330
331
# Return generic error response
332
return JSONResponse(
333
status_code=500,
334
content={
335
"error": "Internal Server Error",
336
"message": "An unexpected error occurred",
337
"timestamp": datetime.utcnow().isoformat()
338
}
339
)
340
341
@app.exception_handler(HTTPException)
342
async def http_exception_handler(request: Request, exc: HTTPException):
343
# Log HTTP exceptions
344
logger.warning(
345
f"HTTP {exc.status_code}: {exc.detail}",
346
extra={
347
"path": request.url.path,
348
"method": request.method,
349
"status_code": exc.status_code
350
}
351
)
352
353
return JSONResponse(
354
status_code=exc.status_code,
355
content={
356
"error": f"HTTP {exc.status_code}",
357
"message": exc.detail,
358
"timestamp": datetime.utcnow().isoformat()
359
},
360
headers=exc.headers
361
)
362
363
@app.get("/error-demo/{error_type}")
364
async def error_demo(error_type: str):
365
if error_type == "404":
366
raise HTTPException(status_code=404, detail="Resource not found")
367
elif error_type == "500":
368
raise Exception("Simulated internal server error")
369
elif error_type == "400":
370
raise HTTPException(status_code=400, detail="Bad request")
371
return {"message": "No error"}
372
```
373
374
### Validation Error Customization
375
376
```python
377
from fastapi import FastAPI, Request
378
from fastapi.exceptions import RequestValidationError
379
from fastapi.responses import JSONResponse
380
from pydantic import BaseModel, validator
381
382
app = FastAPI()
383
384
class ItemModel(BaseModel):
385
name: str
386
price: float
387
description: str = None
388
389
@validator('price')
390
def price_must_be_positive(cls, v):
391
if v <= 0:
392
raise ValueError('Price must be positive')
393
return v
394
395
@validator('name')
396
def name_must_not_be_empty(cls, v):
397
if not v.strip():
398
raise ValueError('Name cannot be empty')
399
return v
400
401
@app.exception_handler(RequestValidationError)
402
async def validation_exception_handler(request: Request, exc: RequestValidationError):
403
errors = []
404
for error in exc.errors():
405
field_path = " -> ".join(str(loc) for loc in error["loc"])
406
errors.append({
407
"field": field_path,
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": "The request contains invalid data",
418
"errors": errors,
419
"total_errors": len(errors)
420
}
421
)
422
423
@app.post("/items/")
424
async def create_item(item: ItemModel):
425
return {"message": "Item created", "item": item}
426
```
427
428
### WebSocket Exception Handling
429
430
```python
431
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
432
from fastapi.exceptions import WebSocketException
433
434
app = FastAPI()
435
436
@app.websocket("/ws")
437
async def websocket_endpoint(websocket: WebSocket):
438
await websocket.accept()
439
try:
440
while True:
441
data = await websocket.receive_text()
442
443
# Validate WebSocket data
444
if not data.strip():
445
raise WebSocketException(
446
code=1003,
447
reason="Empty message not allowed"
448
)
449
450
if len(data) > 1000:
451
raise WebSocketException(
452
code=1009,
453
reason="Message too large"
454
)
455
456
# Echo the message back
457
await websocket.send_text(f"Echo: {data}")
458
459
except WebSocketDisconnect:
460
print("WebSocket client disconnected")
461
except WebSocketException as e:
462
print(f"WebSocket error {e.code}: {e.reason}")
463
await websocket.close(code=e.code, reason=e.reason)
464
except Exception as e:
465
print(f"Unexpected WebSocket error: {e}")
466
await websocket.close(code=1011, reason="Internal server error")
467
```
468
469
### Exception Handling in Dependencies
470
471
```python
472
from fastapi import FastAPI, Depends, HTTPException, Header
473
474
app = FastAPI()
475
476
def verify_auth_header(authorization: str = Header(None)):
477
if not authorization:
478
raise HTTPException(
479
status_code=401,
480
detail="Authorization header missing",
481
headers={"WWW-Authenticate": "Bearer"}
482
)
483
484
if not authorization.startswith("Bearer "):
485
raise HTTPException(
486
status_code=401,
487
detail="Invalid authorization format",
488
headers={"WWW-Authenticate": "Bearer"}
489
)
490
491
token = authorization.replace("Bearer ", "")
492
if token != "valid-token":
493
raise HTTPException(
494
status_code=401,
495
detail="Invalid token",
496
headers={"WWW-Authenticate": "Bearer"}
497
)
498
499
return token
500
501
def get_user_role(token: str = Depends(verify_auth_header)):
502
# Simulate role extraction from token
503
if token == "valid-token":
504
return "admin"
505
return "user"
506
507
def require_admin_role(role: str = Depends(get_user_role)):
508
if role != "admin":
509
raise HTTPException(
510
status_code=403,
511
detail="Admin role required"
512
)
513
return role
514
515
@app.get("/admin/users")
516
async def list_users(role: str = Depends(require_admin_role)):
517
return {"users": ["user1", "user2"], "requesting_role": role}
518
519
@app.get("/profile")
520
async def get_profile(token: str = Depends(verify_auth_header)):
521
return {"message": "Profile data", "token": token}
522
```
523
524
### Context Manager for Exception Handling
525
526
```python
527
from contextlib import asynccontextmanager
528
from fastapi import FastAPI, HTTPException
529
import asyncio
530
531
app = FastAPI()
532
533
@asynccontextmanager
534
async def handle_database_errors():
535
try:
536
yield
537
except ConnectionError:
538
raise HTTPException(
539
status_code=503,
540
detail="Database connection failed"
541
)
542
except TimeoutError:
543
raise HTTPException(
544
status_code=504,
545
detail="Database operation timed out"
546
)
547
except Exception as e:
548
raise HTTPException(
549
status_code=500,
550
detail=f"Database error: {str(e)}"
551
)
552
553
@app.get("/users/{user_id}")
554
async def get_user(user_id: int):
555
async with handle_database_errors():
556
# Simulate database operations
557
if user_id == 1:
558
raise ConnectionError("DB connection lost")
559
elif user_id == 2:
560
await asyncio.sleep(10) # Simulate timeout
561
raise TimeoutError("Operation timed out")
562
elif user_id == 3:
563
raise Exception("Unknown database error")
564
565
return {"user_id": user_id, "name": f"User {user_id}"}
566
```