from fastapi import Request from fastapi.responses import JSONResponse class AppException(Exception): def __init__( self, code: str, message: str, status_code: int = 400, field: str | None = None, headers: dict[str, str] | None = None, ): # Feed the message into Exception.__init__ so ``str(exc)`` # carries it. Previously ``str(AppException(...))`` was the # EMPTY STRING, which meant every call site that flattened a # typed error via ``raise ValueError(str(exc))`` (or logged # ``str(exc)``) produced a blank message — business 404/409s # surfaced as content-free INTERNAL_ERROR 500s over MCP. # (Launch audit 2026-06-10.) super().__init__(message) self.code = code self.message = message self.status_code = status_code self.field = field # Round 4 audit fix-up 2026-05-04: optional response headers so # 429 / 503 paths can carry standard hints (Retry-After) that # were lost when the password lockout was migrated from # ``HTTPException(headers=...)`` to AppException. self.headers = headers class NotFoundError(AppException): def __init__(self, resource: str, resource_id: str | None = None): # Two calling conventions coexist in the codebase: # NotFoundError("Asset", asset_id) → "Asset '' not found" # NotFoundError("post not found") → message used verbatim # The single-arg form previously raised TypeError AT THE RAISE # SITE (missing positional arg), turning every such not-found # path — 25 call sites across feed / asset — into an opaque # 500 instead of the intended 404. Accepting it as a plain # message fixes all of them at once. (Launch audit 2026-06-10.) super().__init__( code="NOT_FOUND", message=( resource if resource_id is None else f"{resource} '{resource_id}' not found" ), status_code=404, ) class ForbiddenError(AppException): def __init__(self, message: str = "Permission denied"): super().__init__(code="FORBIDDEN", message=message, status_code=403) class UnauthorizedError(AppException): def __init__(self, message: str = "Not authenticated"): super().__init__(code="UNAUTHORIZED", message=message, status_code=401) class ProdCloudLockedError(AppException): """Raised when a human tries to mutate a production cloud scope while the global prod-write lock (``cmdb.prod_cloud_writes_enabled``) is engaged. Distinct ``code`` so the frontend can render the dedicated "production locked" explainer instead of a generic permission error. Role-independent: even ``global_admin`` hits this — see :mod:`app.domains.cmdb.prod_write_lock`. """ def __init__(self, message: str): super().__init__( code="PROD_CLOUD_LOCKED", message=message, status_code=403, ) class ConflictError(AppException): def __init__(self, message: str = "Conflict", *, code: str = "CONFLICT"): super().__init__(code=code, message=message, status_code=409) class IdempotencyInProgressError(ConflictError): """A request with this Idempotency-Key is still running. A distinct ``code`` because the client can say something true and specific about THIS 409 — "your previous identical submission is still processing; it will not run twice" — which would be a lie for the ~60 other business conflicts that share the bare ``CONFLICT`` code and the same status. Without it the frontend either shows developer English or mistranslates every other 409. """ def __init__( self, message: str = ( "A request with this Idempotency-Key is already in progress. " "Retry once it completes." ), ): super().__init__(message, code="IDEMPOTENCY_IN_PROGRESS") class IdempotencyKeyInvalidError(AppException): """The client sent an ``Idempotency-Key`` that cannot be stored. The key goes verbatim into ``core_idempotency_records.key`` (``String(128)``), so an oversized key used to surface as a MySQL 1406 DataError → an unhandled 500 on every ``@idempotent`` endpoint. A distinct ``code`` (not the shared ``VALIDATION_ERROR``) because the fix is client-side — regenerate the key — and the frontend can say so. """ def __init__(self, message: str = "Invalid Idempotency-Key header"): super().__init__( code="IDEMPOTENCY_KEY_INVALID", message=message, status_code=422, ) class IdempotencyPayloadMismatchError(AppException): """The same ``Idempotency-Key`` was reused with a DIFFERENT request body. IETF idempotency-key-header semantics: a key identifies ONE operation. Without this check a client bug that reuses a key across different payloads gets the FIRST call's cached success back — a "success" for a write that never happened. 422, not 409: retrying unchanged will never succeed; the client must send a fresh key. """ def __init__( self, message: str = ( "This Idempotency-Key was already used with a different request " "payload. Use a new key for a new operation." ), ): super().__init__( code="IDEMPOTENCY_PAYLOAD_MISMATCH", message=message, status_code=422, ) class ValidationError(AppException): def __init__(self, message: str): super().__init__(code="VALIDATION_ERROR", message=message, status_code=422) class LeaverBlockedError(AppException): """Raised when any code path tries to transition an employee to ``employment_status='left'`` while ``OffboardingBlocker`` rows for that person remain in ``resolution='pending'``. Carries the snapshot of pending blocker IDs so the caller (admin UI, HRIS sync, the Process Advance tick) can present an actionable message: which resources still point at the leaver and need to be transferred or revoked first. HTTP status 409 (Conflict) — the request is structurally valid but the system state forbids it. """ def __init__( self, person_id: str, pending_blocker_ids: list[str], message: str | None = None, audit_id: str | None = None, ): self.person_id = person_id self.pending_blocker_ids = pending_blocker_ids # ``leaver_block_audit`` row id for the denial that produced this # error, when the raising path wrote one. Attribute-only (never in # the user-facing message) so support can jump straight to the # forensic row from a 409 without scanning the audit table by time. self.audit_id = audit_id super().__init__( code="LEAVER_BLOCKED", message=message or ( f"Cannot transition {person_id} to 'left': " f"{len(pending_blocker_ids)} unresolved blocker(s)" ), status_code=409, ) _SAFE_RESPONSE_HEADER_KEYS: frozenset[str] = frozenset({ # Standard hint-only headers safe for any error response. # Round 6 audit fix-up 2026-05-04: allowlist enforced because # ``AppException.headers`` is a free-form dict — without an # allowlist a future caller could pass ``Set-Cookie`` / CORS / # other auth-relevant headers that would bypass the response # middleware's normal sanitisation. New keys must be added here # explicitly after security review. "retry-after", "www-authenticate", "allow", }) def format_exception_for_log( exc: BaseException, *, max_len: int = 500, include_type: bool = True, ) -> str: """Format an exception for audit / error-column / log-line storage. Captures :class:`AppException`'s structured ``.code`` + ``.message`` fields. Historical note: until the 2026-06-10 launch audit, AppException skipped ``super().__init__(message)``, so ``str(exc)`` returned the EMPTY string and naive audit/error-column writes like ``f"{type(exc).__name__}: {exc}"`` silently recorded the useless 15-char sentinel ``"AppException: "``. ``str(exc)`` now carries the message, but this helper remains the canonical formatter because it also captures ``.code`` (never in ``args``) and bounds the length. The Round-21 (2026-05-13) prod incident saw 14 cloud_access_grant blockers stuck at ``last_error_message="AppException: "`` for days, masking a real RBAC FORBIDDEN behind that empty string. This helper is the single canonical formatter for every audit / blocker / provisioning-task / compliance-incident error-recording site so that the same bug-class cannot reoccur the next time someone wires a new error path. * AppException-derived (``.code``, ``.message``) → ``": : "`` * Standard exceptions with args → ``": "`` * Pathological exceptions where both ``str()`` and ``.message`` are empty → ``": "`` so we never erase the entire signal ``include_type=False`` returns just the body text (useful for UI tooltip columns where the type name is already shown separately). ``max_len`` truncates the final output so multi-KB asyncmy / PendingRollbackError stack-trace dumps don't blow out audit detail JSON columns or blocker.last_error_message (1000-char column). """ app_code = getattr(exc, "code", None) app_message = getattr(exc, "message", None) if app_message: body = f"{app_code}: {app_message}" if app_code else app_message else: body = str(exc) or repr(exc) if include_type: body = f"{type(exc).__name__}: {body}" return body[:max_len] # MySQL / TiDB error numbers that mean the ENTIRE transaction was # rolled back server-side (not just the current statement / savepoint): # 1205 — lock-wait timeout (InnoDB gave up waiting on a row lock) # 1213 — deadlock detected (InnoDB chose us as the victim) # 1305 — SAVEPOINT does not exist (the canonical fingerprint of "the # transaction was aborted out from under SQLAlchemy" — emitted # when begin_nested() tries to RELEASE/ROLLBACK a savepoint that # a prior 1205/1213 already destroyed) # All three leave the session unusable: any further work on it — including # a savepoint-local catch-and-continue — operates on a dead transaction # and snowballs into a 1305 cascade. The only valid response is to abort # the unit of work and re-run on a fresh session. _TXN_ABORTING_DB_ERRNOS = frozenset({1205, 1213, 1305}) # MySQL / TiDB error numbers that mean the CONNECTION died (not just the # transaction): the socket was severed and the session is unusable. # 2006 — "MySQL server has gone away" (server closed the connection) # 2013 — "Lost connection to MySQL server during query" (e.g. the # in-flight connection got RST'd — "connection reset by peer" — # when a TiDB Cloud node recycles / scales / drains under us) # Unlike a lock-wait/deadlock these are not the server choosing us as a # victim; the connection itself is gone, so ``pool_pre_ping`` (a checkout- # time probe) can't catch a mid-query drop. The only valid response for an # idempotent unit of work is to re-run on a brand-new connection. _CONNECTION_LOST_DB_ERRNOS = frozenset({2006, 2013}) def is_connection_lost_db_error(exc: BaseException) -> bool: """True if ``exc`` is a MySQL/TiDB connection-drop (server gone away 2006 / lost connection mid-query 2013). These are transient: a fresh session re-running the same idempotent work succeeds. Kept distinct from :func:`is_transaction_aborting_db_error` because the two answer different questions — savepoint catch boundaries care about whether the *transaction* was rolled back, whereas cron fresh-session retry also wants to recover from the *connection* dying out from under it. """ orig = getattr(exc, "orig", None) # ``args`` can be EMPTY (SQLAlchemy wrapping a bare ``ValueError()`` from # a TypeDecorator, a driver ``InterfaceError()`` with no message), and # ``[...][0]`` on it raises IndexError. That matters more since 2026-08-06: # ``best_effort`` calls this as the FIRST statement of its except handler, # so an IndexError here escapes the one code path whose entire job is to # never fail. _args = getattr(orig, "args", None) or () errno = _args[0] if _args else None return errno in _CONNECTION_LOST_DB_ERRNOS def is_transaction_aborting_db_error(exc: BaseException) -> bool: """True if ``exc`` is a MySQL/TiDB error that rolled the WHOLE transaction back server-side (lock-wait 1205 / deadlock 1213 / orphaned-savepoint 1305). Use this at every ``begin_nested()`` catch boundary to decide whether a failure is savepoint-local (IntegrityError, AppException → safe to catch and continue) or transaction-fatal (these errnos → MUST re-raise so the caller can restart on a fresh session). Swallowing a 1205/1213 inside a per-item savepoint loop is what turns a transient, retryable deadlock into a ``SAVEPOINT ... does not exist`` cascade that silently skips every remaining item in the batch. """ orig = getattr(exc, "orig", None) # ``args`` can be EMPTY (SQLAlchemy wrapping a bare ``ValueError()`` from # a TypeDecorator, a driver ``InterfaceError()`` with no message), and # ``[...][0]`` on it raises IndexError. That matters more since 2026-08-06: # ``best_effort`` calls this as the FIRST statement of its except handler, # so an IndexError here escapes the one code path whose entire job is to # never fail. _args = getattr(orig, "args", None) or () errno = _args[0] if _args else None return errno in _TXN_ABORTING_DB_ERRNOS async def app_exception_handler(request: Request, exc: AppException) -> JSONResponse: import logging trace_id = getattr(request.state, "trace_id", None) safe_headers: dict[str, str] | None = None if exc.headers: safe_headers = {} for k, v in exc.headers.items(): if k.lower() in _SAFE_RESPONSE_HEADER_KEYS: safe_headers[k] = v else: logging.getLogger(__name__).error( "AppException header %r dropped — not in allowlist", k, ) if not safe_headers: safe_headers = None return JSONResponse( status_code=exc.status_code, headers=safe_headers, content={ "success": False, "data": None, "meta": {"trace_id": trace_id}, "errors": [ { "code": exc.code, "message": exc.message, **({"field": exc.field} if exc.field else {}), } ], }, )