File size: 9,587 Bytes
e391a84 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | """
interface/api/error_handlers.py
────────────────────────────────
Centralised FastAPI exception handlers.
Every domain exception is mapped to a specific HTTP status code and a
consistent JSON error response format. Registering all handlers in one place
means route handlers stay clean — they simply raise domain exceptions and
let this module translate them to HTTP responses.
Error response format (all errors):
{
"error": "<snake_case_error_code>",
"message": "<human-readable description>",
"context": { ... }, // domain context dict (may be empty)
"timestamp": "2026-05-31T..."
}
HTTP Status Code Mapping:
400 — InvalidSignalError (bad input data)
404 — EntityNotFoundError (resource not found)
409 — ConflictError (duplicate / unique violation)
422 — PredictionOutOfRangeError (model output unprocessable)
422 — PreprocessingError (signal processing failed)
500 — unhandled Exception (unknown / programming error)
503 — DatabaseError (DB unreachable)
503 — BrokerError (message broker unreachable)
503 — ModelInferenceError (AI model failed)
"""
from __future__ import annotations
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from src.domain.exceptions.domain_exceptions import (
ConflictError,
DatabaseError,
DomainException,
EntityNotFoundError,
InvalidSignalError,
PredictionOutOfRangeError,
)
from src.domain.exceptions.pipeline_exceptions import (
BrokerError,
ModelInferenceError,
PreprocessingError,
)
from src.shared.logger import get_logger
logger = get_logger(__name__)
# ── Shared response builder ───────────────────────────────────────────────────
def _error_response(
status_code: int,
error_code: str,
message: str,
context: dict | None = None,
) -> JSONResponse:
"""Build a consistent JSON error payload."""
return JSONResponse(
status_code=status_code,
content={
"error": error_code,
"message": message,
"context": context or {},
"timestamp": datetime.now(timezone.utc).isoformat(),
},
)
# ── Individual exception handlers ─────────────────────────────────────────────
async def handle_invalid_signal(request: Request, exc: InvalidSignalError) -> JSONResponse:
"""400 — PPG signal fails domain validation."""
logger.warning("InvalidSignalError [%s %s]: %s", request.method, request.url.path, exc.message)
return _error_response(400, "invalid_signal", exc.message, exc.context)
async def handle_entity_not_found(request: Request, exc: EntityNotFoundError) -> JSONResponse:
"""404 — Requested entity does not exist in the database."""
logger.info("EntityNotFoundError [%s %s]: %s", request.method, request.url.path, exc.message)
return _error_response(404, "not_found", exc.message, exc.context)
async def handle_conflict(request: Request, exc: ConflictError) -> JSONResponse:
"""409 — Unique constraint / duplicate record violation."""
logger.warning("ConflictError [%s %s]: %s", request.method, request.url.path, exc.message)
return _error_response(409, "conflict", exc.message, exc.context)
async def handle_prediction_out_of_range(
request: Request, exc: PredictionOutOfRangeError
) -> JSONResponse:
"""422 — Model prediction is physiologically implausible."""
logger.warning(
"PredictionOutOfRangeError [%s %s]: %s", request.method, request.url.path, exc.message
)
return _error_response(422, "prediction_out_of_range", exc.message, exc.context)
async def handle_preprocessing_error(request: Request, exc: PreprocessingError) -> JSONResponse:
"""422 — Signal preprocessing failed (bad data or algorithm error)."""
logger.warning(
"PreprocessingError [%s %s]: %s", request.method, request.url.path, exc.message
)
return _error_response(422, "preprocessing_failed", exc.message, exc.context)
async def handle_database_error(request: Request, exc: DatabaseError) -> JSONResponse:
"""503 — Database unavailable or unrecoverable query error."""
logger.error("DatabaseError [%s %s]: %s", request.method, request.url.path, exc.message)
return _error_response(
503,
"database_unavailable",
"The database is temporarily unavailable. Please try again later.",
exc.context,
)
async def handle_broker_error(request: Request, exc: BrokerError) -> JSONResponse:
"""503 — Message broker (RabbitMQ) unreachable or operation failed."""
logger.error("BrokerError [%s %s]: %s", request.method, request.url.path, exc.message)
return _error_response(
503,
"broker_unavailable",
"The message broker is temporarily unavailable. Your data was saved but not queued.",
exc.context,
)
async def handle_model_inference_error(request: Request, exc: ModelInferenceError) -> JSONResponse:
"""503 — AI model inference failed."""
logger.error(
"ModelInferenceError [%s %s]: %s", request.method, request.url.path, exc.message
)
return _error_response(
503,
"model_inference_failed",
f"The AI model '{exc.model_name}' failed to process the request. "
"Please try again later.",
exc.context,
)
async def handle_domain_exception(request: Request, exc: DomainException) -> JSONResponse:
"""400 — Catch-all for any unclassified domain exception."""
logger.warning(
"DomainException [%s %s]: %s", request.method, request.url.path, exc.message
)
return _error_response(400, "domain_error", exc.message, exc.context)
async def handle_unhandled_exception(request: Request, exc: Exception) -> JSONResponse:
"""500 — Completely unexpected / programming error."""
logger.error(
"Unhandled exception [%s %s]: %s",
request.method,
request.url.path,
exc,
exc_info=True,
)
return _error_response(
500,
"internal_error",
"An unexpected internal error occurred. Please contact support.",
)
# ── Registration helper ───────────────────────────────────────────────────────
def register_exception_handlers(app: FastAPI) -> None:
"""
Register all domain exception handlers on a FastAPI application instance.
Call this once inside ``create_app()`` after the app is instantiated.
Order matters: more-specific subclasses must be registered BEFORE their
base classes so FastAPI routes to the correct handler.
"""
# ── 400 ──────────────────────────────────────────────────────────────────
app.add_exception_handler(InvalidSignalError, handle_invalid_signal) # type: ignore
# ── 404 ──────────────────────────────────────────────────────────────────
app.add_exception_handler(EntityNotFoundError, handle_entity_not_found) # type: ignore
# ── 409 ──────────────────────────────────────────────────────────────────
app.add_exception_handler(ConflictError, handle_conflict) # type: ignore
# ── 422 ──────────────────────────────────────────────────────────────────
app.add_exception_handler(PredictionOutOfRangeError, handle_prediction_out_of_range) # type: ignore
app.add_exception_handler(PreprocessingError, handle_preprocessing_error) # type: ignore
# ── 503 ──────────────────────────────────────────────────────────────────
app.add_exception_handler(DatabaseError, handle_database_error) # type: ignore
app.add_exception_handler(BrokerError, handle_broker_error) # type: ignore
app.add_exception_handler(ModelInferenceError, handle_model_inference_error) # type: ignore
# ── 400 catchall (must be AFTER all subclasses) ───────────────────────────
app.add_exception_handler(DomainException, handle_domain_exception) # type: ignore
# ── 500 catchall ─────────────────────────────────────────────────────────
app.add_exception_handler(Exception, handle_unhandled_exception) # type: ignore
logger.info("Exception handlers registered (%d handlers).", 9)
|