Логирование исключений в FastAPI — WARNING / ERROR, один раз, на edge

Опирается на правила: R-ERR-LOG-1 … R-ERR-LOG-4 и R-ERR-LOG-X1 … R-ERR-LOG-X2 из Error Handling Style Guide → раздел 4. Логирование исключений.

Важно знать

DomainError — WARNING в edge-handler. Это ожидаемая ошибка (бизнес-правило сработало), не баг сервиса.

IntegrationError — WARNING при одиночном сбое (CB ещё закрыт). ERROR когда CB открылся — это инцидент.

TechnicalError и catch-all Exception — ERROR с полным stacktrace и структурным контекстом.

Логируем один раз — на edge. Не на каждом уровне call stack — одна ошибка = одна запись в логах.

logger.error(...); raise — двойное логирование, запрещено. Либо логируй и обработай, либо проброс без лога.

logger.error(str(exc)) без exc_info — теряется stacktrace. Использовать logger.exception(...) или logger.error(..., exc_info=exc).

Логгер — structlog с JSON-форматом в проде, контекст через contextvars.

Логирование исключения — не «на всякий случай», это источник правды для оператора и алёрта. Перелогирование (одна ошибка в трёх местах) создаёт шум; недолог (catch без лога) скрывает проблему. Уровень — это семантика для алёртов: WARNING = «ожидаемо, в норме»; ERROR = «надо смотреть». Раскрытие правил R-ERR-LOG-* ниже.

structlog как логгер

В FastAPI-сервисах UCP-стека используем structlog — структурированный JSON-логгер с поддержкой contextvars.

# app/logging_config.py
import logging
import structlog

def configure_logging(json_logs: bool = True) -> None:
    shared_processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
    ]
    if json_logs:
        processors = shared_processors + [structlog.processors.JSONRenderer()]
    else:
        processors = shared_processors + [structlog.dev.ConsoleRenderer()]

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
    )

Контекст (request_id, user_id, trace_id) прописывается в middleware через structlog.contextvars.bind_contextvars:

# app/middleware.py
import uuid
import structlog
from fastapi import Request

async def logging_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        method=request.method,
        path=request.url.path,
    )
    response = await call_next(request)
    return response

После этого любой logger.warning(...) в edge-handler автоматически несёт request_id, method, path без явной передачи.

DomainError — WARNING

R-ERR-LOG-1: бизнес-правило сработало. Это ожидаемая ситуация, не баг.

logger = structlog.get_logger()

async def _handle_domain(request: Request, exc: DomainError) -> JSONResponse:
    logger.warning(
        "domain rule violated",
        exc_type=type(exc).__name__,
        error=str(exc),
    )
    ...

Почему не ERROR:

Ложные алёрты. ERROR-алёрты обычно настроены «звонить дежурному». Если каждый InsufficientFundsError шлёт уведомление в 3 часа ночи, команда быстро отключит алёрты.
DomainError — это нормально. Пользователь нажал «Оплатить», средств не хватило — ожидаемая ветка флоу, не сбой. Лог нужен для аналитики (сколько раз сегодня сработало это правило), но не для тревоги.
Метрики разделяют. На WARNING-исключения смотрят через app_errors_total{type="domain"}. Резкий рост = бизнес-условие изменилось незаметно. Стабильный baseline = всё нормально.

Для конкретных DomainError-наследников, которые имеют структурированные поля, structlog-вызов добавляет их явно:

async def _handle_insufficient_funds(
    request: Request, exc: InsufficientFundsError
) -> JSONResponse:
    logger.warning(
        "domain rule violated",
        exc_type="InsufficientFundsError",
        customer_id=exc.customer_id,
        requested=str(exc.requested),
        available=str(exc.available),
    )
    ...

Это даёт JSON-запись с отдельными полями — поиск по customer_id в Loki / CloudWatch Insights без парсинга строки.

IntegrationError — WARNING или ERROR

R-ERR-LOG-2: зависит от состояния circuit breaker.

WARNING — одиночный сбой, CB ещё закрыт. Пара 5xx в потоке нормальной нагрузки — норма (деплой внешки, сетевые флуктуации):

async def _handle_payment_gateway(
    request: Request, exc: PaymentGatewayError
) -> JSONResponse:
    logger.warning(
        "integration error",
        exc_type=type(exc).__name__,
        error=str(exc),
    )
    ...

ERROR — CB открылся. Это уже инцидент: внешка деградировала достаточно, чтобы наш CB защитил нагрузку. Фиксируем отдельным типом:

async def _handle_payment_unavailable(
    request: Request, exc: PaymentGatewayUnavailableError
) -> JSONResponse:
    logger.error(
        "payment gateway circuit breaker open",
        exc_type="PaymentGatewayUnavailableError",
    )
    ...

На практике алёрт «CB открыт» лучше делать через метрику (aiobreaker / custom Counter), а не через лог-агрегатор — метрика быстрее и стабильнее.

TechnicalError и catch-all — ERROR

R-ERR-LOG-3: внутренняя проблема — полный stacktrace и контекст.

async def _handle_technical(request: Request, exc: TechnicalError) -> JSONResponse:
    logger.error(
        "technical error",
        exc_info=exc,                           # ← structlog включит stacktrace
        exc_type=type(exc).__name__,
    )
    ...


async def _handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
    logger.error(
        "unexpected error — not in hierarchy",
        exc_info=exc,
        exc_type=type(exc).__name__,
    )
    ...

exc_info=exc в structlog — аналог exc_info=True в stdlib logging. Передаёт объект исключения, structlog включает traceback в JSON-запись. Альтернатива — logger.exception(...) внутри except-блока (неявно захватывает текущее исключение):

try:
    ...
except Exception:
    logger.exception("unexpected error")
    raise

Но в handler'е FastAPI мы уже вне except-контекста — используем exc_info=exc явно.

Логируем один раз — на edge

R-ERR-LOG-4: исключение логируется в register_error_handlers, и больше нигде.

# ПЛОХО — двойное логирование

# core/use_cases/cancel_order.py
async def handle(self, cmd: CancelOrderCommand) -> Order:
    try:
        order = await self._orders.get_by_id(cmd.order_id)
        order.cancel(cmd.reason)
        await self._orders.save(order)
        return order
    except DomainError as e:
        logger.warning("domain error in cancel", error=str(e))  # ← лог №1
        raise                                                     # ← edge залогирует №2

# app/error_handlers.py
async def _handle_domain(request: Request, exc: DomainError) -> JSONResponse:
    logger.warning("domain rule violated", error=str(exc))       # ← лог №2
    ...

Что не так:

Одна ошибка = два лога разного качества (один без structlog-контекста из middleware, второй с ним). Оператор путается.
Логи в двух местах с разным количеством полей — поиск по request_id даёт два результата на одно событие.
Метрика app_errors_total инкрементируется один раз, логов два — несоответствие смущает.

Принцип: исключение проходит насквозь до edge без логирования. Edge — единственное место с полным structlog-контекстом из middleware (request_id, path, method + всё через bind_contextvars).

Что запрещено

Антипаттерн	Правило	Что взамен
`logger.error("failed", exc_info=exc); raise exc`	`R-ERR-LOG-X1`	Либо логируй и обработай, либо `raise` без лога
`logger.error(str(exc))` без `exc_info`	`R-ERR-LOG-X2`	`logger.error("context", exc_info=exc)`
`logger.exception("failed")` внутри handler (двойной лог)	`R-ERR-LOG-X1`	Логировать только на edge
`print(traceback.format_exc())`	`R-ERR-LOG-X2`	Использовать structlog с `exc_info`
WARNING на `TechnicalError`	`R-ERR-LOG-3`	ERROR — это не ожидаемая ошибка
ERROR на каждый `DomainError`	`R-ERR-LOG-1`	WARNING — ожидаемая, не баг

R-ERR-LOG-X1 подробно:

# ПЛОХО — в out-adapter
async def register(self, cmd: RegisterCommand) -> RegisterResult:
    try:
        resp = await self._client.post("/register", json=_to_api(cmd))
        resp.raise_for_status()
        return _to_domain(resp.json())
    except httpx.HTTPStatusError as e:
        logger.error("sber 5xx", exc_info=e)          # ← лог №1 здесь
        raise PaymentGatewayError("sber 5xx") from e  # ← edge залогирует №2

В out-adapter — только mapping в port-specific exception, без лога. Лог — на edge.

R-ERR-LOG-X2 подробно:

# ПЛОХО — stacktrace потерян
logger.error(f"unexpected: {exc}")

# ПЛОХО — stacktrace потерян по-другому
logger.error("unexpected error", message=str(exc))

# ХОРОШО
logger.error("unexpected error", exc_info=exc, exc_type=type(exc).__name__)

str(exc) — только message без traceback. В JSON-логе будет строка, без стека вызовов. При отладке производственных инцидентов stacktrace — основное, что нужно.

Куда дальше

Где raise, где catch — почему ноль try/except в handler/service.
Mapping в ProblemDetails — что ещё делает edge-handler помимо лога.
Observability ошибок — метрики, trace, алёрты.
Иерархия исключений — какие типы соответствуют каким уровням.
Retry-семантика — IntegrationError и CB.
Result types vs exceptions — когда Result допустим.
Error Handling Style Guide → раздел 4 — нормативные формулировки.