Error Handling Skill
Expert structured error handling for FastAPI backends and React/Next.js frontends with consistent error messages and logging.
Quick Reference
| Pattern | Backend | Frontend |
|---|---|---|
| Custom exception | class FeeNotPaidError(AppException) | N/A |
| Try-catch | try: ... except SpecificError: | try { } catch (e) { } |
| Global handler | @app.exception_handler | ErrorBoundary component |
| User message | detail field in response | Toast/Snackbar |
| Error logging | logger.error(...) | Console/Sentry |
Custom Exceptions (Backend)
Base Exception Hierarchy
# backend/app/errors/exceptions.py
from fastapi import HTTPException
from typing import Any
class AppException(HTTPException):
"""Base application exception with user-friendly messages."""
def __init__(
self,
status_code: int,
detail: str,
user_message: str,
headers: dict[str, Any] | None = None,
):
self.user_message = user_message
super().__init__(status_code=status_code, detail=detail, headers=headers)
class NotFoundError(AppException):
"""Resource not found (404)."""
def __init__(self, resource: str, identifier: str):
super().__init__(
status_code=404,
detail=f"{resource} with id '{identifier}' not found",
user_message=f"{resource} not found. Please check and try again.",
)
class ValidationError(AppException):
"""Validation failed (400)."""
def __init__(self, field: str, reason: str):
super().__init__(
status_code=400,
detail=f"Validation error for field '{field}': {reason}",
user_message=f"Invalid value for {field}. {reason}",
)
class UnauthorizedError(AppException):
"""Authentication required (401)."""
def __init__(self, reason: str = "Authentication required"):
super().__init__(
status_code=401,
detail=reason,
user_message="Please log in to continue.",
headers={"WWW-Authenticate": "Bearer"},
)
class ForbiddenError(AppException):
"""Permission denied (403)."""
def __init__(self, action: str):
super().__init__(
status_code=403,
detail=f"Permission denied for action: {action}",
user_message=f"You don't have permission to {action}.",
)
class ConflictError(AppException):
"""Resource conflict (409)."""
def __init__(self, resource: str, reason: str):
super().__init__(
status_code=409,
detail=f"Conflict for {resource}: {reason}",
user_message=f"{resource} conflict. {reason}",
)
class RateLimitError(AppException):
"""Too many requests (429)."""
def __init__(self, retry_after: int = 60):
super().__init__(
status_code=429,
detail=f"Rate limit exceeded. Retry after {retry_after} seconds.",
user_message="Too many requests. Please wait a moment and try again.",
headers={"Retry-After": str(retry_after)},
)
Domain-Specific Exceptions
# backend/app/errors/domains.py
from .exceptions import NotFoundError, ValidationError, ConflictError
class StudentNotFoundError(NotFoundError):
def __init__(self, student_id: int):
super().__init__(resource="Student", identifier=str(student_id))
class FeeNotPaidError(ConflictError):
def __init__(self, student_id: int, amount_due: float):
super().__init__(
resource="Fee",
reason=f"Student {student_id} has unpaid fee of ${amount_due:.2f}",
)
class AttendanceAlreadyMarkedError(ConflictError):
def __init__(self, student_id: int, date: str):
super().__init__(
resource="Attendance",
reason=f"Attendance already marked for student {student_id} on {date}",
)
class InvalidGradeError(ValidationError):
def __init__(self, grade: str, valid_grades: list[str]):
super().__init__(
field="grade",
reason=f"'{grade}' is not valid. Must be one of: {', '.join(valid_grades)}",
)
class InsufficientBalanceError(ValidationError):
def __init__(self, required: float, available: float):
super().__init__(
field="amount",
reason=f"Insufficient balance. Required: ${required:.2f}, Available: ${available:.2f}",
)
Try-Catch Patterns
Narrow Try Blocks
# GOOD: Specific exception, narrow scope
try:
student = await get_student_by_id(student_id)
except StudentNotFoundError:
raise StudentNotFoundError(student_id)
# BAD: Too broad, catches everything
try:
student = await get_student_by_id(student_id)
calculate_fees(student)
send_notification(student)
update_records(student)
except Exception:
pass # Swallowed!
Cleanup with Finally
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def database_connection():
conn = await get_db_connection()
try:
yield conn
except Exception as e:
await conn.rollback()
raise
finally:
await conn.close()
async def transfer_funds(from_account: int, to_account: int, amount: float):
async with database_connection() as conn:
try:
await conn.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", amount, from_account)
await conn.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", amount, to_account)
await conn.commit()
except InsufficientBalanceError:
await conn.rollback()
raise
Error Logging (Backend)
Structured JSON Logging
# backend/app/core/logger.py
import logging
import json
from datetime import datetime
from typing import Any
from contextvars import ContextVar
import traceback
# Correlation ID for request tracing
correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")
class StructuredLogger:
"""Structured JSON logger with correlation IDs."""
def __init__(self, name: str):
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.INFO)
def _log_record(self, level: str, message: str, extra: dict[str, Any] = None):
record = {
"timestamp": datetime.utcnow().isoformat(),
"level": level,
"message": message,
"correlation_id": correlation_id_var.get(),
**(extra or {}),
}
return json.dumps(record)
def info(self, message: str, **extra):
self.logger.info(self._log_record("INFO", message, extra))
def warning(self, message: str, **extra):
self.logger.warning(self._log_record("WARNING", message, extra))
def error(self, message: str, error: Exception = None, **extra):
log_extra = {**extra}
if error:
log_extra["error_type"] = type(error).__name__
log_extra["error_message"] = str(error)
log_extra["stack_trace"] = traceback.format_exc()
self.logger.error(self._log_record("ERROR", message, log_extra))
logger = StructuredLogger(__name__)
Using the Logger
from app.core.logger import logger, correlation_id_var
async def get_student(student_id: int) -> Student:
logger.info("Fetching student", student_id=student_id)
try:
student = await db.get_student(student_id)
logger.info("Student fetched successfully", student_id=student_id, has_fees=bool(student.fees))
return student
except StudentNotFoundError:
logger.warning("Student not found", student_id=student_id)
raise
except Exception as e:
logger.error("Unexpected error fetching student", error=e, student_id=student_id)
raise # Re-raise for handler
Global Error Handler (Backend)