HTTPException, кастомные исключения, error handlers
Правильная обработка ошибок делает API понятным и безопасным. В этой теме вы научитесь возвращать HTTP-ошибки, создавать кастомные исключения и обрабатывать ошибки глобально.
Самый простой способ вернуть ошибку — HTTPException:
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get('/items/{item_id}')
def get_item(item_id: int):
item = get_item_from_db(item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return itemHTTPExceptionОтвет клиенту:
{
"detail": "Item not found"
}HTTPException(
status_code=404, # Обязательный: HTTP-статус
detail="Not found", # Обязательный: сообщение
headers={"X-Error": "ID"} # Необязательный: заголовки
)Используйте status для читаемости:
from fastapi import HTTPException, status
# 400 Bad Request
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid request"
)
# 401 Unauthorized
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated"
)
# 403 Forbidden
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
# 404 Not Found
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Resource not found"
)
# 409 Conflict
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Resource already exists"
)
# 422 Unprocessable Entity
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Validation error"
)
# 500 Internal Server Error
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Internal error"
)Можно вернуть структурированную ошибку:
from fastapi import HTTPException
class ValidationErrorDetail(BaseModel):
field: str
message: str
@app.post('/users')
def create_user(user: UserCreate):
# Проверка уникальности email
if email_exists(user.email):
raise HTTPException(
status_code=409,
detail={
"error": "email_conflict",
"message": "Email already registered",
"field": "email"
}
)Ответ:
{
"detail": {
"error": "email_conflict",
"message": "Email already registered",
"field": "email"
}
}Создайте базовый класс для ваших исключений:
from fastapi import HTTPException, status
class AppException(HTTPException):
"""Базовое исключение приложения"""
def __init__(
self,
status_code: int,
error_code: str,
message: str,
details: dict | None = None
):
super().__init__(
status_code=status_code,
detail={
"error_code": error_code,
"message": message,
"details": details or {}
}
)
# Специализированные исключения
class NotFoundException(AppException):
def __init__(self, resource: str, resource_id: int):
super().__init__(
status_code=status.HTTP_404_NOT_FOUND,
error_code="RESOURCE_NOT_FOUND",
message=f"{resource} with id {resource_id} not found"
)
class ValidationException(AppException):
def __init__(self, field: str, message: str):
super().__init__(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
error_code="VALIDATION_ERROR",
message=message,
details={"field": field}
)
class UnauthorizedException(AppException):
def __init__(self, message: str = "Not authenticated"):
super().__init__(
status_code=status.HTTP_401_UNAUTHORIZED,
error_code="UNAUTHORIZED",
message=message
)@app.get('/users/{user_id}')
def get_user(user_id: int):
user = db.get_user(user_id)
if not user:
raise NotFoundException("User", user_id)
return user
@app.post('/users')
def create_user(user: UserCreate):
if not is_valid_email(user.email):
raise ValidationException("email", "Invalid email format")
if email_exists(user.email):
raise ValidationException("email", "Email already registered")
return create_user_in_db(user)Ответ при ошибке:
{
"detail": {
"error_code": "RESOURCE_NOT_FOUND",
"message": "User with id 123 not found",
"details": {}
}
}Можно перехватывать все исключения одного типа:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={
"error_code": exc.detail["error_code"],
"message": exc.detail["message"],
"details": exc.detail["details"],
"path": str(request.url)
}
)Теперь все AppException будут обрабатываться этим handler'ом.
Можно перехватывать и стандартные исключения Python:
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
return JSONResponse(
status_code=400,
content={"detail": f"Invalid value: {str(exc)}"}
)
@app.exception_handler(KeyError)
async def key_error_handler(request: Request, exc: KeyError):
return JSONResponse(
status_code=400,
content={"detail": f"Missing key: {str(exc)}"}
)from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(404)
async def not_found_handler(request: Request, exc):
return JSONResponse(
status_code=404,
content={
"error_code": "ROUTE_NOT_FOUND",
"message": f"Route {request.url} not found"
}
)from fastapi import FastAPI, HTTPException, status, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from enum import Enum
app = FastAPI()
# Типы ошибок
class ErrorType(str, Enum):
VALIDATION = "VALIDATION_ERROR"
NOT_FOUND = "NOT_FOUND"
UNAUTHORIZED = "UNAUTHORIZED"
FORBIDDEN = "FORBIDDEN"
CONFLICT = "CONFLICT"
INTERNAL = "INTERNAL_ERROR"
# Модель ошибки
class ErrorDetail(BaseModel):
error_type: ErrorType
message: str
field: str | None = None
path: str
# Базовое исключение
class APIException(HTTPException):
def __init__(
self,
status_code: int,
error_type: ErrorType,
message: str,
field: str | None = None
):
super().__init__(
status_code=status_code,
detail={
"error_type": error_type.value,
"message": message,
"field": field
}
)
# Специализированные исключения
class ValidationException(APIException):
def __init__(self, message: str, field: str | None = None):
super().__init__(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
error_type=ErrorType.VALIDATION,
message=message,
field=field
)
class NotFoundException(APIException):
def __init__(self, resource: str):
super().__init__(
status_code=status.HTTP_404_NOT_FOUND,
error_type=ErrorType.NOT_FOUND,
message=f"{resource} not found"
)
class UnauthorizedException(APIException):
def __init__(self, message: str = "Not authenticated"):
super().__init__(
status_code=status.HTTP_401_UNAUTHORIZED,
error_type=ErrorType.UNAUTHORIZED,
message=message
)
class ConflictException(APIException):
def __init__(self, message: str):
super().__init__(
status_code=status.HTTP_409_CONFLICT,
error_type=ErrorType.CONFLICT,
message=message
)
# Глобальный обработчик
@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
return JSONResponse(
status_code=exc.status_code,
content={
**exc.detail,
"path": str(request.url)
}
)
# Обработчик стандартных исключений
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
# Логируем ошибку (в реальности — в лог-файл или Sentry)
print(f"Unexpected error: {exc}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"error_type": ErrorType.INTERNAL.value,
"message": "An unexpected error occurred",
"path": str(request.url)
}
)
# Примеры использования
@app.get('/users/{user_id}')
def get_user(user_id: int):
user = get_user_from_db(user_id)
if not user:
raise NotFoundException("User")
return user
@app.post('/users')
def create_user(user: UserCreate):
if not is_valid_email(user.email):
raise ValidationException("Invalid email format", "email")
if email_exists(user.email):
raise ConflictException("Email already registered")
return create_user_in_db(user)
@app.get('/admin')
def get_admin_data():
# Проверка аутентификации
raise UnauthorizedException("Admin access required")Запрос /users/999 (несуществующий пользователь):
{
"error_type": "NOT_FOUND",
"message": "User not found",
"field": null,
"path": "http://localhost:8000/users/999"
}Pydantic автоматически возвращает ошибки валидации:
class UserCreate(BaseModel):
email: str
age: int = Field(..., ge=18)
@app.post('/users')
def create_user(user: UserCreate):
...Запрос с {"email": "invalid", "age": 15} вернёт:
{
"detail": [
{
"type": "value_error",
"loc": ["body", "email"],
"msg": "value is not a valid email address",
"input": "invalid"
},
{
"type": "greater_than_equal",
"loc": ["body", "age"],
"msg": "Input should be greater than or equal to 18",
"input": 15,
"ctx": {"ge": 18}
}
]
}@app.get('/items/{id}')
def get_item(id: int):
if not item_exists(id):
return {"error": "Not found"} # Ошибка: вернёт 200 OK!
return get_item(id)Проблема: Клиент получит 200 OK с ошибкой в теле.
Решение:
@app.get('/items/{id}')
def get_item(id: int):
if not item_exists(id):
raise HTTPException(status_code=404, detail="Item not found")
return get_item(id)@app.exception_handler(Exception)
async def handle_all(request: Request, exc: Exception):
return JSONResponse(status_code=500, content={"detail": "Error"})Проблема: Все ошибки (включая 404, 422) вернут 500.
Решение: Обрабатывайте специфичные исключения отдельно:
@app.exception_handler(HTTPException)
async def http_handler(request: Request, exc: HTTPException):
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
@app.exception_handler(Exception)
async def general_handler(request: Request, exc: Exception):
# Только неожиданные ошибки
return JSONResponse(status_code=500, content={"detail": "Internal error"})@app.exception_handler(Exception)
async def handle(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"detail": str(exc), "traceback": traceback.format_exc()}
)Проблема: Клиент увидит стек трейс, SQL-запросы, пути к файлам.
Решение:
@app.exception_handler(Exception)
async def handle(request: Request, exc: Exception):
# Логируем детали внутри
logger.error(f"Error: {exc}\n{traceback.format_exc()}")
# Клиенту — общая информация
return JSONResponse(
status_code=500,
content={"detail": "An internal error occurred"}
)Лаборатория
«Доменные исключения и кастомные handlers»
доступна в публичном GitLab. Она проверяет иерархию кастомных исключений,
@app.exception_handler, структурированные ошибки с error_code и
сокрытие внутренних деталей от клиента.
Склонируйте репозиторий и запустите тесты из его корня:
git clone --branch v1.0 --depth 1 https://gitlab.potapov.me/courses/fastapi_pro.git
cd fastapi_pro
uv sync --group test
uv run --group test pytest error_handling/testsДалее: Dependency Injection