pypi-fastapi

1
# Exception Handling
2

3
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.
4

5
## Capabilities
6

7
### HTTP Exception
8

9
The primary exception class for handling HTTP errors with proper status codes and response formatting.
10

11
```python { .api }
12
class HTTPException(Exception):
13
    def __init__(
14
        self,
15
        status_code: int,
16
        detail: Any = None,
17
        headers: Dict[str, str] = None
18
    ) -> None:
19
        """
20
        HTTP exception with status code and detail message.
21
        
22
        Parameters:
23
        - status_code: HTTP status code (400, 401, 404, 500, etc.)
24
        - detail: Error detail message or structured data
25
        - headers: Additional HTTP headers to include in error response
26
        """
27
        self.status_code = status_code
28
        self.detail = detail
29
        self.headers = headers
30
```
31

32
### WebSocket Exception
33

34
Exception class specifically for WebSocket connection errors.
35

36
```python { .api }
37
class WebSocketException(Exception):
38
    def __init__(
39
        self,
40
        code: int,
41
        reason: str = None
42
    ) -> None:
43
        """
44
        WebSocket error exception.
45
        
46
        Parameters:
47
        - code: WebSocket close code (1000-4999)
48
        - reason: Human-readable close reason
49
        """
50
        self.code = code
51
        self.reason = reason
52
```
53

54
### FastAPI Error
55

56
Generic runtime error exception for FastAPI framework issues.
57

58
```python { .api }
59
class FastAPIError(Exception):
60
    def __init__(self, message: str = None) -> None:
61
        """
62
        Generic FastAPI runtime error.
63
        
64
        Parameters:
65
        - message: Error message describing the issue
66
        """
67
        self.message = message
68
```
69

70
### Validation Exceptions
71

72
Exception classes for handling request and response validation errors with detailed error information.
73

74
```python { .api }
75
class ValidationException(Exception):
76
    def __init__(self, errors: List[dict]) -> None:
77
        """
78
        Base validation error exception.
79
        
80
        Parameters:
81
        - errors: List of validation error dictionaries
82
        """
83
        self.errors = errors
84

85
class RequestValidationError(ValidationException):
86
    def __init__(self, errors: List[ErrorWrapper]) -> None:
87
        """
88
        Request validation error with detailed error information.
89
        
90
        Parameters:
91
        - errors: List of Pydantic ErrorWrapper objects containing
92
                 field-specific validation error details
93
        """
94
        self.errors = errors
95
        self.body = getattr(errors, "body", None)
96

97
class WebSocketRequestValidationError(ValidationException):
98
    def __init__(self, errors: List[ErrorWrapper]) -> None:
99
        """
100
        WebSocket request validation error.
101
        
102
        Parameters:
103
        - errors: List of Pydantic ErrorWrapper objects for WebSocket
104
                 parameter validation failures
105
        """
106
        self.errors = errors
107

108
class ResponseValidationError(ValidationException):
109
    def __init__(self, errors: List[ErrorWrapper]) -> None:
110
        """
111
        Response validation error for response model validation failures.
112
        
113
        Parameters:
114
        - errors: List of Pydantic ErrorWrapper objects containing
115
                 response validation error details
116
        """
117
        self.errors = errors
118
```
119

120
### Exception Handler Decorator
121

122
Decorator method for registering custom exception handlers on FastAPI and APIRouter instances.
123

124
```python { .api }
125
def exception_handler(
126
    self,
127
    exc_class_or_status_code: Union[int, Type[Exception]]
128
) -> Callable[[Callable], Callable]:
129
    """
130
    Decorator for adding exception handlers.
131

132
    Parameters:
133
    - exc_class_or_status_code: Exception class or HTTP status code to handle
134

135
    Returns:
136
    Decorator function for exception handler registration
137
    """
138
```
139

140
### Exception Handler Function Type
141

142
Type signature for custom exception handler functions.
143

144
```python { .api }
145
async def exception_handler_function(
146
    request: Request,
147
    exc: Exception
148
) -> Response:
149
    """
150
    Exception handler function signature.
151
    
152
    Parameters:
153
    - request: HTTP request object
154
    - exc: Exception instance that was raised
155
    
156
    Returns:
157
    Response object to send to client
158
    """
159
```
160

161
## Usage Examples
162

163
### Basic HTTP Exception Handling
164

165
```python
166
from fastapi import FastAPI, HTTPException, status
167

168
app = FastAPI()
169

170
@app.get("/items/{item_id}")
171
async def read_item(item_id: int):
172
    if item_id == 0:
173
        raise HTTPException(
174
            status_code=status.HTTP_404_NOT_FOUND,
175
            detail="Item not found"
176
        )
177
    if item_id < 0:
178
        raise HTTPException(
179
            status_code=status.HTTP_400_BAD_REQUEST,
180
            detail="Item ID must be positive",
181
            headers={"X-Error": "Invalid item ID"}
182
        )
183
    return {"item_id": item_id, "name": f"Item {item_id}"}
184
```
185

186
### Custom Exception Handler
187

188
```python
189
from fastapi import FastAPI, Request, HTTPException
190
from fastapi.responses import JSONResponse
191
from fastapi.exceptions import RequestValidationError
192
from starlette.exceptions import HTTPException as StarletteHTTPException
193

194
app = FastAPI()
195

196
@app.exception_handler(StarletteHTTPException)
197
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
198
    return JSONResponse(
199
        status_code=exc.status_code,
200
        content={
201
            "error": "HTTP Exception",
202
            "message": exc.detail,
203
            "path": request.url.path,
204
            "method": request.method
205
        }
206
    )
207

208
@app.exception_handler(RequestValidationError)
209
async def validation_exception_handler(request: Request, exc: RequestValidationError):
210
    return JSONResponse(
211
        status_code=422,
212
        content={
213
            "error": "Validation Error",
214
            "message": "Invalid request data",
215
            "details": exc.errors(),
216
            "body": exc.body
217
        }
218
    )
219

220
@app.get("/items/{item_id}")
221
async def read_item(item_id: int):
222
    if item_id == 42:
223
        raise HTTPException(status_code=418, detail="I'm a teapot")
224
    return {"item_id": item_id}
225
```
226

227
### Custom Exception Classes
228

229
```python
230
from fastapi import FastAPI, Request, HTTPException
231
from fastapi.responses import JSONResponse
232

233
app = FastAPI()
234

235
# Custom exception classes
236
class ItemNotFoundError(Exception):
237
    def __init__(self, item_id: int):
238
        self.item_id = item_id
239
        self.message = f"Item with ID {item_id} not found"
240
        super().__init__(self.message)
241

242
class InsufficientPermissionsError(Exception):
243
    def __init__(self, required_role: str, user_role: str):
244
        self.required_role = required_role
245
        self.user_role = user_role
246
        self.message = f"Required role: {required_role}, user role: {user_role}"
247
        super().__init__(self.message)
248

249
class BusinessLogicError(Exception):
250
    def __init__(self, message: str, error_code: str):
251
        self.message = message
252
        self.error_code = error_code
253
        super().__init__(self.message)
254

255
# Exception handlers
256
@app.exception_handler(ItemNotFoundError)
257
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
258
    return JSONResponse(
259
        status_code=404,
260
        content={
261
            "error": "Item Not Found",
262
            "message": exc.message,
263
            "item_id": exc.item_id
264
        }
265
    )
266

267
@app.exception_handler(InsufficientPermissionsError)
268
async def insufficient_permissions_handler(request: Request, exc: InsufficientPermissionsError):
269
    return JSONResponse(
270
        status_code=403,
271
        content={
272
            "error": "Insufficient Permissions",
273
            "message": exc.message,
274
            "required_role": exc.required_role,
275
            "user_role": exc.user_role
276
        }
277
    )
278

279
@app.exception_handler(BusinessLogicError)
280
async def business_logic_handler(request: Request, exc: BusinessLogicError):
281
    return JSONResponse(
282
        status_code=422,
283
        content={
284
            "error": "Business Logic Error",
285
            "message": exc.message,
286
            "error_code": exc.error_code
287
        }
288
    )
289

290
# Routes using custom exceptions
291
@app.get("/items/{item_id}")
292
async def get_item(item_id: int):
293
    if item_id == 999:
294
        raise ItemNotFoundError(item_id)
295
    return {"item_id": item_id, "name": f"Item {item_id}"}
296

297
@app.delete("/items/{item_id}")
298
async def delete_item(item_id: int, user_role: str = "user"):
299
    if user_role != "admin":
300
        raise InsufficientPermissionsError("admin", user_role)
301
    if item_id <= 0:
302
        raise BusinessLogicError("Cannot delete items with non-positive IDs", "INVALID_ID")
303
    return {"message": f"Item {item_id} deleted"}
304
```
305

306
### Exception Handling with Logging
307

308
```python
309
import logging
310
from datetime import datetime
311
from fastapi import FastAPI, Request, HTTPException
312
from fastapi.responses import JSONResponse
313

314
# Configure logging
315
logging.basicConfig(level=logging.INFO)
316
logger = logging.getLogger(__name__)
317

318
app = FastAPI()
319

320
@app.exception_handler(Exception)
321
async def global_exception_handler(request: Request, exc: Exception):
322
    # Log the exception
323
    logger.error(
324
        f"Unhandled exception: {type(exc).__name__}: {str(exc)}",
325
        extra={
326
            "path": request.url.path,
327
            "method": request.method,
328
            "timestamp": datetime.utcnow().isoformat()
329
        }
330
    )
331
    
332
    # Return generic error response
333
    return JSONResponse(
334
        status_code=500,
335
        content={
336
            "error": "Internal Server Error",
337
            "message": "An unexpected error occurred",
338
            "timestamp": datetime.utcnow().isoformat()
339
        }
340
    )
341

342
@app.exception_handler(HTTPException)
343
async def http_exception_handler(request: Request, exc: HTTPException):
344
    # Log HTTP exceptions
345
    logger.warning(
346
        f"HTTP {exc.status_code}: {exc.detail}",
347
        extra={
348
            "path": request.url.path,
349
            "method": request.method,
350
            "status_code": exc.status_code
351
        }
352
    )
353
    
354
    return JSONResponse(
355
        status_code=exc.status_code,
356
        content={
357
            "error": f"HTTP {exc.status_code}",
358
            "message": exc.detail,
359
            "timestamp": datetime.utcnow().isoformat()
360
        },
361
        headers=exc.headers
362
    )
363

364
@app.get("/error-demo/{error_type}")
365
async def error_demo(error_type: str):
366
    if error_type == "404":
367
        raise HTTPException(status_code=404, detail="Resource not found")
368
    elif error_type == "500":
369
        raise Exception("Simulated internal server error")
370
    elif error_type == "400":
371
        raise HTTPException(status_code=400, detail="Bad request")
372
    return {"message": "No error"}
373
```
374

375
### Validation Error Customization
376

377
```python
378
from fastapi import FastAPI, Request
379
from fastapi.exceptions import RequestValidationError
380
from fastapi.responses import JSONResponse
381
from pydantic import BaseModel, validator
382

383
app = FastAPI()
384

385
class ItemModel(BaseModel):
386
    name: str
387
    price: float
388
    description: str = None
389
    
390
    @validator('price')
391
    def price_must_be_positive(cls, v):
392
        if v <= 0:
393
            raise ValueError('Price must be positive')
394
        return v
395
    
396
    @validator('name')
397
    def name_must_not_be_empty(cls, v):
398
        if not v.strip():
399
            raise ValueError('Name cannot be empty')
400
        return v
401

402
@app.exception_handler(RequestValidationError)
403
async def validation_exception_handler(request: Request, exc: RequestValidationError):
404
    errors = []
405
    for error in exc.errors():
406
        field_path = " -> ".join(str(loc) for loc in error["loc"])
407
        errors.append({
408
            "field": field_path,
409
            "message": error["msg"],
410
            "type": error["type"],
411
            "input": error.get("input")
412
        })
413
    
414
    return JSONResponse(
415
        status_code=422,
416
        content={
417
            "error": "Validation Failed",
418
            "message": "The request contains invalid data",
419
            "errors": errors,
420
            "total_errors": len(errors)
421
        }
422
    )
423

424
@app.post("/items/")
425
async def create_item(item: ItemModel):
426
    return {"message": "Item created", "item": item}
427
```
428

429
### WebSocket Exception Handling
430

431
```python
432
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
433
from fastapi.exceptions import WebSocketException
434

435
app = FastAPI()
436

437
@app.websocket("/ws")
438
async def websocket_endpoint(websocket: WebSocket):
439
    await websocket.accept()
440
    try:
441
        while True:
442
            data = await websocket.receive_text()
443
            
444
            # Validate WebSocket data
445
            if not data.strip():
446
                raise WebSocketException(
447
                    code=1003,
448
                    reason="Empty message not allowed"
449
                )
450
            
451
            if len(data) > 1000:
452
                raise WebSocketException(
453
                    code=1009,
454
                    reason="Message too large"
455
                )
456
            
457
            # Echo the message back
458
            await websocket.send_text(f"Echo: {data}")
459
            
460
    except WebSocketDisconnect:
461
        print("WebSocket client disconnected")
462
    except WebSocketException as e:
463
        print(f"WebSocket error {e.code}: {e.reason}")
464
        await websocket.close(code=e.code, reason=e.reason)
465
    except Exception as e:
466
        print(f"Unexpected WebSocket error: {e}")
467
        await websocket.close(code=1011, reason="Internal server error")
468
```
469

470
### Exception Handling in Dependencies
471

472
```python
473
from fastapi import FastAPI, Depends, HTTPException, Header
474

475
app = FastAPI()
476

477
def verify_auth_header(authorization: str = Header(None)):
478
    if not authorization:
479
        raise HTTPException(
480
            status_code=401,
481
            detail="Authorization header missing",
482
            headers={"WWW-Authenticate": "Bearer"}
483
        )
484
    
485
    if not authorization.startswith("Bearer "):
486
        raise HTTPException(
487
            status_code=401,
488
            detail="Invalid authorization format",
489
            headers={"WWW-Authenticate": "Bearer"}
490
        )
491
    
492
    token = authorization.replace("Bearer ", "")
493
    if token != "valid-token":
494
        raise HTTPException(
495
            status_code=401,
496
            detail="Invalid token",
497
            headers={"WWW-Authenticate": "Bearer"}
498
        )
499
    
500
    return token
501

502
def get_user_role(token: str = Depends(verify_auth_header)):
503
    # Simulate role extraction from token
504
    if token == "valid-token":
505
        return "admin"
506
    return "user"
507

508
def require_admin_role(role: str = Depends(get_user_role)):
509
    if role != "admin":
510
        raise HTTPException(
511
            status_code=403,
512
            detail="Admin role required"
513
        )
514
    return role
515

516
@app.get("/admin/users")
517
async def list_users(role: str = Depends(require_admin_role)):
518
    return {"users": ["user1", "user2"], "requesting_role": role}
519

520
@app.get("/profile")
521
async def get_profile(token: str = Depends(verify_auth_header)):
522
    return {"message": "Profile data", "token": token}
523
```
524

525
### Context Manager for Exception Handling
526

527
```python
528
from contextlib import asynccontextmanager
529
from fastapi import FastAPI, HTTPException
530
import asyncio
531

532
app = FastAPI()
533

534
@asynccontextmanager
535
async def handle_database_errors():
536
    try:
537
        yield
538
    except ConnectionError:
539
        raise HTTPException(
540
            status_code=503,
541
            detail="Database connection failed"
542
        )
543
    except TimeoutError:
544
        raise HTTPException(
545
            status_code=504,
546
            detail="Database operation timed out"
547
        )
548
    except Exception as e:
549
        raise HTTPException(
550
            status_code=500,
551
            detail=f"Database error: {str(e)}"
552
        )
553

554
@app.get("/users/{user_id}")
555
async def get_user(user_id: int):
556
    async with handle_database_errors():
557
        # Simulate database operations
558
        if user_id == 1:
559
            raise ConnectionError("DB connection lost")
560
        elif user_id == 2:
561
            await asyncio.sleep(10)  # Simulate timeout
562
            raise TimeoutError("Operation timed out")
563
        elif user_id == 3:
564
            raise Exception("Unknown database error")
565
        
566
        return {"user_id": user_id, "name": f"User {user_id}"}
567
```