Обработка ошибок

В этом разделе описывается система обработки ошибок в Boilerplate. Вы узнаете о стандартных ошибках FastAPI, пользовательских исключениях и о том, как создать единообразную обработку ошибок в вашем API.

Обзор обработки ошибок

FastAPI предоставляет встроенные механизмы для обработки различных типов ошибок:

Ошибки валидации данных — когда входные данные не соответствуют ожидаемым схемам
HTTP-исключения — для возврата соответствующих HTTP-статусов и сообщений об ошибках
Обработчики исключений — глобальные или специфичные обработчики для разных типов ошибок

Стандартные HTTP-исключения

FastAPI предоставляет класс HTTPException для возврата HTTP-ошибок:

python

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=400, 
            detail="ID элемента не может быть отрицательным"
        )
    
    item = await get_item(item_id)
    if item is None:
        raise HTTPException(
            status_code=404,
            detail=f"Элемент с ID {item_id} не найден",
            headers={"X-Error": "Item not found"}  # Дополнительные заголовки
        )
        
    return item

Кастомные исключения

В Boilerplate определены пользовательские исключения для унификации обработки ошибок:

python

# app/core/exceptions.py
from typing import Any, Dict, Optional

from fastapi import HTTPException, status

class NotFoundError(HTTPException):
    """
    Исключение для ошибки "Не найдено" (HTTP 404)
    """
    def __init__(
        self, 
        detail: str = "Запрашиваемый ресурс не найден", 
        headers: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=detail,
            headers=headers
        )

class BadRequestError(HTTPException):
    """
    Исключение для ошибки "Неверный запрос" (HTTP 400)
    """
    def __init__(
        self, 
        detail: str = "Неверный запрос", 
        headers: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=detail,
            headers=headers
        )

class UnauthorizedError(HTTPException):
    """
    Исключение для ошибки "Неавторизован" (HTTP 401)
    """
    def __init__(
        self, 
        detail: str = "Не авторизован", 
        headers: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=detail,
            headers=headers or {"WWW-Authenticate": "Bearer"}
        )

class ForbiddenError(HTTPException):
    """
    Исключение для ошибки "Доступ запрещен" (HTTP 403)
    """
    def __init__(
        self, 
        detail: str = "Нет прав доступа", 
        headers: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=detail,
            headers=headers
        )

class ConflictError(HTTPException):
    """
    Исключение для ошибки "Конфликт" (HTTP 409)
    """
    def __init__(
        self, 
        detail: str = "Конфликт с текущим состоянием ресурса", 
        headers: Optional[Dict[str, Any]] = None
    ):
        super().__init__(
            status_code=status.HTTP_409_CONFLICT,
            detail=detail,
            headers=headers
        )

Использование кастомных исключений в коде:

python

# app/api/routes/users.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app import crud
from app.api import deps
from app.core.exceptions import NotFoundError, ConflictError

router = APIRouter()

@router.get("/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(deps.get_db),
):
    user = await crud.user.get(db, id=user_id)
    if not user:
        raise NotFoundError(detail=f"Пользователь с ID {user_id} не найден")
    return user

@router.post("/")
async def create_user(
    user_in: schemas.UserCreate,
    db: AsyncSession = Depends(deps.get_db),
):
    existing_user = await crud.user.get_by_email(db, email=user_in.email)
    if existing_user:
        raise ConflictError(detail="Пользователь с таким email уже существует")
    
    # Создание пользователя
    return await crud.user.create(db, obj_in=user_in)

Глобальные обработчики исключений

FastAPI позволяет определить глобальные обработчики для разных типов исключений. Это позволяет унифицировать формат ответов с ошибками:

python

# app/core/exception_handlers.py
from typing import Any, Dict, Union

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from sqlalchemy.exc import SQLAlchemyError
from starlette.exceptions import HTTPException

from app.core.config import settings

def add_exception_handlers(app: FastAPI) -> None:
    """
    Добавляет глобальные обработчики исключений к FastAPI приложению.
    """
    app.add_exception_handler(
        HTTPException, http_exception_handler
    )
    app.add_exception_handler(
        RequestValidationError, validation_exception_handler
    )
    app.add_exception_handler(
        SQLAlchemyError, sqlalchemy_exception_handler
    )
    app.add_exception_handler(
        Exception, general_exception_handler
    )

async def http_exception_handler(
    request: Request, exc: HTTPException
) -> JSONResponse:
    """
    Обработчик для HTTP исключений.
    """
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": {
                "code": exc.status_code,
                "message": exc.detail,
                "type": "http_error"
            }
        },
        headers=exc.headers,
    )

async def validation_exception_handler(
    request: Request, exc: Union[RequestValidationError, ValidationError]
) -> JSONResponse:
    """
    Обработчик для ошибок валидации.
    """
    errors = []
    for error in exc.errors():
        error_info = {
            "loc": error.get("loc", []),
            "msg": error.get("msg", ""),
            "type": error.get("type", "")
        }
        errors.append(error_info)
    
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error": {
                "code": 422,
                "message": "Ошибка валидации данных",
                "type": "validation_error",
                "details": errors
            }
        },
    )

async def sqlalchemy_exception_handler(
    request: Request, exc: SQLAlchemyError
) -> JSONResponse:
    """
    Обработчик для ошибок базы данных.
    """
    error_message = str(exc)
    
    # В продакшене скрываем детали SQL ошибок
    if not settings.DEBUG:
        error_message = "Ошибка базы данных"
    
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": {
                "code": 500,
                "message": error_message,
                "type": "database_error"
            }
        },
    )

async def general_exception_handler(
    request: Request, exc: Exception
) -> JSONResponse:
    """
    Обработчик для всех остальных типов исключений.
    """
    error_message = str(exc)
    
    # В продакшене скрываем детали внутренних ошибок
    if not settings.DEBUG:
        error_message = "Внутренняя ошибка сервера"
    
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": {
                "code": 500,
                "message": error_message,
                "type": "server_error"
            }
        },
    )

Регистрация обработчиков исключений в приложении:

python

# app/main.py
from fastapi import FastAPI

from app.core.exception_handlers import add_exception_handlers
from app.api.api import api_router

app = FastAPI(
    title="Boilerplate API",
    description="Документация API для Boilerplate",
    version="0.1.0",
)

# Добавляем обработчики исключений
add_exception_handlers(app)

# Регистрация маршрутов API
app.include_router(api_router, prefix="/api/v1")