Coverage for fastapi_restly / _exception_handlers.py: 89%
70 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-24 11:13 +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.configure`
12and as a fallback by :func:`fastapi_restly.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 psycopg's ``orig.pgcode``) and the SQLite
19error-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 Looks at ``orig.pgcode`` (set by psycopg / psycopg2 / asyncpg-via-psycopg)
65 and, when available, ``orig.diag.constraint_name`` /
66 ``orig.diag.column_name`` to enrich the message.
67 """
68 pgcode = getattr(orig, "pgcode", None)
69 if not pgcode:
70 return None
72 base = _PG_SQLSTATE_DETAILS.get(pgcode)
73 if base is None: 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true
74 return None
76 # ``diag`` is a psycopg-specific attribute holding fielded error info.
77 diag = getattr(orig, "diag", None)
78 constraint_name = getattr(diag, "constraint_name", None) if diag else None
79 column_name = getattr(diag, "column_name", None) if diag else None
81 if pgcode == "23505" and constraint_name:
82 return f"{base}: {constraint_name}"
83 if pgcode == "23503" and constraint_name: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 return f"{base}: {constraint_name}"
85 if pgcode == "23502" and column_name:
86 return f"{base} on column {column_name!r}"
87 if pgcode == "23514" and constraint_name: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true
88 return f"{base}: {constraint_name}"
89 return base
92# Mapping from SQLite error-message prefixes to a clean detail message.
93# SQLite's IntegrityError.args[0] (and ``str(orig)``) follow predictable
94# patterns, e.g. ``"UNIQUE constraint failed: user.username"``.
95_SQLITE_PREFIX_DETAILS: tuple[tuple[str, str], ...] = (
96 ("UNIQUE constraint failed:", "Unique constraint violated"),
97 ("FOREIGN KEY constraint failed", "Foreign key constraint violated"),
98 ("NOT NULL constraint failed:", "Not-null constraint violated"),
99 ("CHECK constraint failed:", "Check constraint violated"),
100 ("PRIMARY KEY must be unique", "Unique constraint violated (primary key)"),
101)
104def _extract_sqlite_detail(orig: Any) -> str | None:
105 """Return a user-facing detail message for a SQLite-driver error."""
106 text = str(orig).strip()
107 if not text: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 return None
110 for prefix, base in _SQLITE_PREFIX_DETAILS:
111 if not text.startswith(prefix):
112 continue
113 # Try to surface the column / constraint info that SQLite tacks on
114 # after the colon. ``UNIQUE constraint failed: user.username`` →
115 # ``"Unique constraint violated on user.username"``.
116 remainder = text[len(prefix) :].strip().lstrip(":").strip()
117 if remainder:
118 return f"{base} on {remainder}"
119 return base
120 return None
123def _build_integrity_detail(exc: IntegrityError) -> str:
124 """Build a clean HTTP 409 detail message from a SQLAlchemy IntegrityError.
126 Best-effort across dialects:
128 * PostgreSQL — switches on ``exc.orig.pgcode`` (SQLSTATE class 23).
129 * SQLite — pattern-matches ``str(exc.orig)`` against known prefixes.
130 * Anything else — returns a generic fallback that includes a truncated
131 copy of the original error text so the body is still useful for
132 debugging without being huge.
133 """
134 orig = getattr(exc, "orig", None)
135 if orig is not None: 135 ↛ 146line 135 didn't jump to line 146 because the condition on line 135 was always true
136 pg_detail = _extract_postgres_detail(orig)
137 if pg_detail is not None:
138 return pg_detail
140 sqlite_detail = _extract_sqlite_detail(orig)
141 if sqlite_detail is not None:
142 return sqlite_detail
144 # Generic fallback. Prefer the original driver error text (it's usually
145 # the most informative); truncate so we don't dump a giant SQL statement.
146 raw = str(orig) if orig is not None else str(exc)
147 raw = raw.strip()
148 if len(raw) > _MAX_ORIG_TEXT_LENGTH:
149 raw = raw[:_MAX_ORIG_TEXT_LENGTH] + "...(truncated)"
151 base = "Conflict with current state of the resource"
152 if raw: 152 ↛ 154line 152 didn't jump to line 154 because the condition on line 152 was always true
153 return f"{base}: {raw}"
154 return base
157# ---------------------------------------------------------------------------
158# The handler & registration helper
159# ---------------------------------------------------------------------------
162def integrity_error_handler(request: Request, exc: Exception) -> JSONResponse:
163 """Translate a SQLAlchemy IntegrityError into HTTP 409 Conflict.
165 Signature accepts ``Exception`` instead of ``IntegrityError`` to satisfy
166 Starlette's exception-handler typing; runtime code narrows it.
167 """
168 assert isinstance(exc, IntegrityError) # noqa: S101 - registered for IntegrityError only
169 detail = _build_integrity_detail(exc)
170 return JSONResponse(status_code=409, content={"detail": detail})
173def register_default_exception_handlers(app: FastAPI) -> None:
174 """Idempotently install fastapi-restly default exception handlers on ``app``.
176 * Skips if a handler for :class:`IntegrityError` is already registered on
177 ``app`` — we always defer to the user.
178 * Skips if we have already installed handlers on this ``app`` instance
179 (so calling from both :func:`fastapi_restly.configure` and
180 :func:`fastapi_restly.include_view` is safe).
181 """
182 if getattr(app.state, _HANDLERS_INSTALLED_FLAG, False):
183 return
185 # Respect a user-registered handler if one is already in place.
186 if IntegrityError in app.exception_handlers:
187 setattr(app.state, _HANDLERS_INSTALLED_FLAG, True)
188 return
190 app.add_exception_handler(IntegrityError, integrity_error_handler)
191 setattr(app.state, _HANDLERS_INSTALLED_FLAG, True)
194__all__ = ["integrity_error_handler", "register_default_exception_handlers"]