arm-core/api/src/core/exception_handlers.py

"""Обработчики исключений приложения: единый формат ответа `{"detail": ...}`.

`IntegrityError` → 409: для известных уникальных ограничений — доменный 409 с кодом из
`common/constraint_conflicts.py` (гонка pre-check/INSERT), иначе общая строка; прочие `SQLAlchemyError` → 500.
Ошибки БД пишутся в лог одной записью уровня ERROR без текста исключения: `str()` ошибки SQLAlchemy/asyncpg содержит
параметры запроса и DETAIL PostgreSQL со значениями ключей (email, ИНН) — это персональные данные. В лог идут путь и
метод запроса, классы ошибок, SQLSTATE, код разрешённого конфликта, текст SQL с плейсхолдерами (без значений, усечён) и
стек вызовов без сообщения исключения.
"""

import traceback

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from loguru import logger
from sqlalchemy.exc import DBAPIError, IntegrityError, SQLAlchemyError

from src.common.constraint_conflicts import resolve_unique_conflict
from src.common.exceptions import AuthenticationFailedError, DomainError
from src.services.session_audit import SessionAudit, request_client

LOG_STATEMENT_LIMIT = 200  # символов SQL-текста (с плейсхолдерами) в записи лога


def log_db_error(request: Request, exc: SQLAlchemyError, message: str, *, resolved_code: str | None = None) -> None:
    """Одна запись ERROR об ошибке БД при обработке запроса — без параметров запроса и текста ошибки драйвера.

    `resolved_code` — код доменного конфликта, в который отображена гонка уникальных ограничений (для диагностики).
    """
    orig = exc.orig if isinstance(exc, DBAPIError) else None
    statement = exc.statement if isinstance(exc, DBAPIError) else None
    logger.error(
        message + " path={path}, method={method}, error_type={error_type}, db_error_type={db_error_type}, "
        "sqlstate={sqlstate}, resolved_code={resolved_code}, statement={statement}\n{stack}",
        path=request.url.path,
        method=request.method,
        error_type=type(exc).__name__,
        db_error_type=type(orig).__name__ if orig is not None else None,
        sqlstate=getattr(orig, "sqlstate", None),
        resolved_code=resolved_code,
        statement=repr(statement[:LOG_STATEMENT_LIMIT]) if statement else None,
        stack="".join(traceback.format_tb(exc.__traceback__)).rstrip(),
    )


def register_exception_handlers(app: FastAPI) -> None:
    # Доменные ошибки — до обработчиков `SQLAlchemyError`/`Exception` (spec directories-api: формат detail-объекта)
    @app.exception_handler(DomainError)
    async def domain_error_handler(_: Request, exc: DomainError):
        return JSONResponse(
            status_code=exc.status,
            content={"detail": {"code": exc.code, "message": exc.message, **jsonable_encoder(exc.extra)}},
        )

    @app.exception_handler(RequestValidationError)
    async def request_validation_exception_handler(_: Request, exc: RequestValidationError):
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            # ctx ошибки может содержать исключение (ValueError из валидатора) — сериализуем через jsonable_encoder
            content={"detail": jsonable_encoder(exc.errors())},
        )

    @app.exception_handler(IntegrityError)
    async def integrity_error_handler(request: Request, exc: IntegrityError):
        # Гонка pre-check/INSERT по известному уникальному ограничению → тот же 409 с кодом, что отдал бы pre-check
        conflict = resolve_unique_conflict(exc)
        log_db_error(
            request,
            exc,
            "Нарушение целостности данных в БД при обработке запроса.",
            resolved_code=conflict.code if conflict is not None else None,
        )
        if conflict is not None:
            return JSONResponse(
                status_code=status.HTTP_409_CONFLICT,
                content={"detail": {"code": conflict.code, "message": conflict.message}},
            )
        return JSONResponse(
            status_code=status.HTTP_409_CONFLICT,
            content={"detail": "нарушение целостности данных в БД"},
        )

    @app.exception_handler(SQLAlchemyError)
    async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
        log_db_error(request, exc, "Ошибка базы данных при обработке запроса.")
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"detail": "ошибка базы данных"},
        )

    @app.exception_handler(AuthenticationFailedError)
    async def authentication_failed_handler(request: Request, exc: AuthenticationFailedError):
        # Обработчик работает после отката транзакции запроса: LOGIN_FAILED пишется в отдельной (OQ-64). Приложение
        # без `state.session_audit` (тестовые мини-приложения) отвечает так же, но без записи в журнал
        session_audit = getattr(request.app.state, "session_audit", None)
        if isinstance(session_audit, SessionAudit):
            ip_address, user_agent = request_client(request)
            await session_audit.login_failed(
                exc.user_id, exc.audit_description, ip_address=ip_address, user_agent=user_agent
            )
        return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}, headers=exc.headers)

    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
        # Не ломаем существующие ответы endpoint-ов и прокидываем исходный detail.
        return JSONResponse(
            status_code=exc.status_code,
            content={"detail": exc.detail},
            headers=exc.headers,
        )

    @app.exception_handler(Exception)
    async def unhandled_exception_handler(_: Request, exc: Exception):
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"detail": "внутренняя ошибка сервера"},
        )