Coverage for fastapi_restly / _exception_handlers.py: 91%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-09 18:41 +0000
1"""Default FastAPI exception handlers installed by fastapi-restly.
3Currently this module provides a translation layer from SQLAlchemy
4:class:`~sqlalchemy.exc.IntegrityError` (unique-constraint, foreign-key,
5not-null, and check-constraint violations) into a clean HTTP 409 Conflict
6response. Without this handler, an ``IntegrityError`` bubbles up to FastAPI
7and turns into a 500 Internal Server Error, which is misleading for clients
8(the server is fine; the request conflicts with the current state of the
9resource).
11The handler is installed automatically by :func:`~fastapi_restly.db.configure`
12and as a fallback by :func:`~fastapi_restly.views.include_view`. Users can opt out
13by calling ``fr.configure(install_default_exception_handlers=False)`` or by
14registering their own handler for ``IntegrityError`` *before* the framework
15gets a chance to install one.
17The detail-message extraction is best-effort: it understands the most common
18PostgreSQL SQLSTATE codes (via ``orig.sqlstate`` / ``orig.pgcode``) and the
19SQLite error-message conventions. For unrecognised dialects/messages we fall back
20to a generic conflict message that includes a truncated version of the
21underlying error text.
22"""
24from __future__ import annotations
26from typing import Any
28from fastapi import FastAPI, Request
29from fastapi.responses import JSONResponse
30from sqlalchemy.exc import IntegrityError
32# Maximum length of original-error text we are willing to echo back. Keeps
33# response bodies sane and avoids accidentally leaking long SQL strings.
34_MAX_ORIG_TEXT_LENGTH = 500
36# Marker stored on ``app.state`` so we know we've already installed our
37# handlers on this FastAPI instance. Public so it is easy to inspect from
38# tests or user code.
39_HANDLERS_INSTALLED_FLAG = "_fr_default_exception_handlers_installed"
42# ---------------------------------------------------------------------------
43# Detail extraction
44# ---------------------------------------------------------------------------
47# PostgreSQL SQLSTATE codes — see
48# https://www.postgresql.org/docs/current/errcodes-appendix.html (class 23
49# "Integrity Constraint Violation").
50_PG_SQLSTATE_DETAILS: dict[str, str] = {
51 "23505": "Unique constraint violated",
52 "23503": "Foreign key constraint violated",
53 "23502": "Not-null constraint violated",
54 "23514": "Check constraint violated",
55 "23000": "Integrity constraint violated",
56 "23001": "Restrict violation",
57 "23P01": "Exclusion constraint violated",
58}
61def _extract_postgres_detail(orig: Any) -> str | None:
62 """Return a user-facing detail message for a Postgres-driver error.
64 Reads the SQLSTATE code and, when psycopg's ``orig.diag`` is present, its
65 ``constraint_name`` / ``column_name`` to enrich the message. The attribute
66 holding the code differs by driver: psycopg 3 exposes ``sqlstate``, psycopg 2
67 exposes ``pgcode`` -- read whichever is present. Reading only ``pgcode``
68 silently degraded every psycopg 3 error to the generic fallback.
69 """
70 pgcode = getattr(orig, "sqlstate", None) or getattr(orig, "pgcode", None)
71 if not pgcode:
72 return None
74 base = _PG_SQLSTATE_DETAILS.get(pgcode)
75 if base is None: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true
76 return None
78 # ``diag`` is a psycopg-specific attribute holding fielded error info.
79 diag = getattr(orig, "diag", None)
80 constraint_name = getattr(diag, "constraint_name", None) if diag else None
81 column_name = getattr(diag, "column_name", None) if diag else None
83 if pgcode == "23505" and constraint_name:
84 return f"{base}: {constraint_name}"
85 if pgcode == "23503" and constraint_name: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true
86 return f"{base}: {constraint_name}"
87 if pgcode == "23502" and column_name:
88 return f"{base} on column {column_name!r}"
89 if pgcode == "23514" and constraint_name: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 return f"{base}: {constraint_name}"
91 return base
94# Mapping from SQLite error-message prefixes to a clean detail message.
95# SQLite's IntegrityError.args[0] (and ``str(orig)``) follow predictable
96# patterns, e.g. ``"UNIQUE constraint failed: user.username"``.
97_SQLITE_PREFIX_DETAILS: tuple[tuple[str, str], ...] = (
98 ("UNIQUE constraint failed:", "Unique constraint violated"),
99 ("FOREIGN KEY constraint failed", "Foreign key constraint violated"),
100 ("NOT NULL constraint failed:", "Not-null constraint violated"),
101 ("CHECK constraint failed:", "Check constraint violated"),
102 ("PRIMARY KEY must be unique", "Unique constraint violated (primary key)"),
103)
106def _extract_sqlite_detail(orig: Any) -> str | None:
107 """Return a user-facing detail message for a SQLite-driver error."""
108 text = str(orig).strip()
109 if not text: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 return None
112 for prefix, base in _SQLITE_PREFIX_DETAILS:
113 if not text.startswith(prefix):
114 continue
115 # Try to surface the column / constraint info that SQLite tacks on
116 # after the colon. ``UNIQUE constraint failed: user.username`` →
117 # ``"Unique constraint violated on user.username"``.
118 remainder = text[len(prefix) :].strip().lstrip(":").strip()
119 if remainder:
120 return f"{base} on {remainder}"
121 return base
122 return None
125def _build_integrity_detail(exc: IntegrityError) -> str:
126 """Build a clean HTTP 409 detail message from a SQLAlchemy IntegrityError.
128 Best-effort across dialects:
130 * PostgreSQL — switches on the SQLSTATE code (``exc.orig.sqlstate`` /
131 ``pgcode``, class 23).
132 * SQLite — pattern-matches ``str(exc.orig)`` against known prefixes.
133 * Anything else — returns a generic fallback that includes a truncated
134 copy of the original error text so the body is still useful for
135 debugging without being huge.
136 """
137 orig = getattr(exc, "orig", None)
138 if orig is not None: 138 ↛ 149line 138 didn't jump to line 149 because the condition on line 138 was always true
139 pg_detail = _extract_postgres_detail(orig)
140 if pg_detail is not None:
141 return pg_detail
143 sqlite_detail = _extract_sqlite_detail(orig)
144 if sqlite_detail is not None:
145 return sqlite_detail
147 # Generic fallback. Prefer the original driver error text (it's usually
148 # the most informative); truncate so we don't dump a giant SQL statement.
149 raw = str(orig) if orig is not None else str(exc)
150 raw = raw.strip()
151 if len(raw) > _MAX_ORIG_TEXT_LENGTH:
152 raw = raw[:_MAX_ORIG_TEXT_LENGTH] + "...(truncated)"
154 base = "Conflict with current state of the resource"
155 if raw: 155 ↛ 157line 155 didn't jump to line 157 because the condition on line 155 was always true
156 return f"{base}: {raw}"
157 return base
160# ---------------------------------------------------------------------------
161# The handler & registration helper
162# ---------------------------------------------------------------------------
165def integrity_error_handler(request: Request, exc: Exception) -> JSONResponse:
166 """Translate a SQLAlchemy IntegrityError into HTTP 409 Conflict.
168 Signature accepts ``Exception`` instead of ``IntegrityError`` to satisfy
169 Starlette's exception-handler typing; runtime code narrows it.
170 """
171 assert isinstance(exc, IntegrityError) # noqa: S101 - registered for IntegrityError only
172 detail = _build_integrity_detail(exc)
173 return JSONResponse(status_code=409, content={"detail": detail})
176def register_default_exception_handlers(app: FastAPI) -> None:
177 """Idempotently install fastapi-restly default exception handlers on ``app``.
179 * Skips if a handler for :class:`~sqlalchemy.exc.IntegrityError` is already
180 registered on ``app`` — we always defer to the user.
181 * Skips if we have already installed handlers on this ``app`` instance
182 (so calling from both :func:`~fastapi_restly.db.configure` and
183 :func:`~fastapi_restly.views.include_view` is safe).
184 """
185 if getattr(app.state, _HANDLERS_INSTALLED_FLAG, False):
186 return
188 # Respect a user-registered handler if one is already in place.
189 if IntegrityError in app.exception_handlers:
190 setattr(app.state, _HANDLERS_INSTALLED_FLAG, True)
191 return
193 app.add_exception_handler(IntegrityError, integrity_error_handler)
194 setattr(app.state, _HANDLERS_INSTALLED_FLAG, True)
195 _register_on_built_stack(app)
198def _register_on_built_stack(app: FastAPI) -> None:
199 """Also register on an application that has already been called.
201 Starlette builds its middleware stack on the first call and the stack caches
202 the handler map, so a view registered after that would leave this handler out
203 and an ``IntegrityError`` would surface as a 500. Entering a test client to
204 run the lifespan is enough to build it.
206 Reach the ``ExceptionMiddleware`` already in the stack rather than discarding
207 the stack to force a rebuild: rebuilding constructs every user middleware
208 again, dropping whatever state the live instances hold, and it would briefly
209 disarm Starlette's own "cannot add middleware after startup" check.
210 """
211 stack = getattr(app, "middleware_stack", None)
212 if stack is None:
213 return
215 from starlette.middleware.exceptions import ExceptionMiddleware
217 node: Any = stack
218 while node is not None:
219 if isinstance(node, ExceptionMiddleware):
220 node.add_exception_handler(IntegrityError, integrity_error_handler)
221 return
222 node = getattr(node, "app", None)
224 import warnings
226 # A middleware that stores its wrapped app under another name ends the walk.
227 # Rare, but the alternative is unexplained 500s on late-registered views.
228 warnings.warn(
229 "Restly could not reach Starlette's ExceptionMiddleware on the built "
230 "middleware stack: a middleware in the chain does not expose its "
231 "wrapped app as `.app`. Views registered after the application first "
232 "ran will surface IntegrityError as a 500 rather than a 409.",
233 RuntimeWarning,
234 stacklevel=2,
235 )
238__all__ = ["integrity_error_handler", "register_default_exception_handlers"]