embeddevio
83d5a23406
- app.py: create_app() собирает мок-сервер и отдаёт /openapi.json из спецификации. - errors.py: ApiError + хендлеры, приводящие 4xx/5xx/422 к схеме Error. - deps.py: Bearer-токен и cookie-сессия (ADR-008). - schemas.py: pydantic-модели запросов, зеркалящие схемы контракта (extra=forbid, writeOnly для секретов).
89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""Единый формат ошибок API: ``{code, message, details}`` (схема Error из контракта)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
STATUS_TO_CODE: dict[int, str] = {
|
|
400: "bad_request",
|
|
401: "unauthorized",
|
|
403: "forbidden",
|
|
404: "not_found",
|
|
409: "conflict",
|
|
422: "validation_failed",
|
|
429: "rate_limited",
|
|
500: "internal_error",
|
|
503: "agent_unavailable",
|
|
}
|
|
|
|
|
|
class ApiError(Exception):
|
|
"""Ошибка API, сериализуемая в схему Error."""
|
|
|
|
def __init__(
|
|
self,
|
|
status_code: int,
|
|
code: str,
|
|
message: str,
|
|
details: dict[str, Any] | None = None,
|
|
) -> None:
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
self.code = code
|
|
self.message = message
|
|
self.details = details
|
|
|
|
|
|
def error_payload(code: str, message: str, details: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
"""Тело ошибки в контрактном виде (``details`` всегда присутствует, может быть null)."""
|
|
return {"code": code, "message": message, "details": details}
|
|
|
|
|
|
def install_error_handlers(app: FastAPI) -> None:
|
|
"""Приводит все ошибки (включая 422 и 500) к схеме Error."""
|
|
|
|
@app.exception_handler(ApiError)
|
|
async def _api_error(_: Request, exc: ApiError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=error_payload(exc.code, exc.message, exc.details),
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def _validation_error(_: Request, exc: RequestValidationError) -> JSONResponse:
|
|
issues = [
|
|
{
|
|
"path": "/" + "/".join(str(part) for part in error.get("loc", ())),
|
|
"code": str(error.get("type", "invalid")),
|
|
"message": str(error.get("msg", "invalid value")),
|
|
}
|
|
for error in exc.errors()
|
|
]
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content=error_payload(
|
|
"validation_failed",
|
|
"Тело запроса не соответствует схеме",
|
|
{"issues": issues},
|
|
),
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def _http_error(_: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
code = STATUS_TO_CODE.get(exc.status_code, "internal_error")
|
|
return JSONResponse(
|
|
status_code=exc.status_code, content=error_payload(code, str(exc.detail), None)
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def _unhandled(_: Request, __: Exception) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content=error_payload("internal_error", "Внутренняя ошибка", None),
|
|
)
|