Подумайте о своём API как о ресторане. Если вдруг на кухне что-то пошло не так — сгорела плита или закончились продукты — но никто не знает, что с этим делать, гости просто останутся голодными и уйдут недовольными. Примерно так же ведёт себя API без нормальной обработки ошибок: случилась беда, а вместо внятного ответа — тишина или странный текст ни о чём.
Наша цель — настроить систему, которая поможет держать всё под контролем. Мы научимся ловить и обрабатывать ошибки аккуратно, выдавать пользователям понятные сообщения, а сами при этом всё фиксировать в логах и сохранять стабильную работу приложения, даже если внутри что-то пошло не так.
Проект: управление задачами (Todo App)
Мы продолжаем развивать наше приложение для управления задачами. На данном этапе ваше API уже может обрабатывать CRUD-операции, поддерживать асинхронность, и имеет базовые обработчики ошибок. Теперь мы внедрим полноценную систему обработки ошибок.
Шаг 1: Создание пользовательских исключений
Пользовательские исключения помогают разделить логику обработки ошибок. Например, нам понадобится исключение для случая, когда задача не найдена.
from fastapi import HTTPException
class TaskNotFoundException(HTTPException):
def __init__(self):
super().__init__(status_code=404, detail="Task not found")
Теперь, если задача отсутствует в базе данных, вместо общего ответа с ошибкой можно использовать это исключение.
Шаг 2: Настройка кастомных обработчиков
FastAPI позволяет создавать кастомные обработчики ошибок для определённых типов исключений. Это удобно для консистентной обработки специфических проблем.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(TaskNotFoundException)
async def task_not_found_handler(request: Request, exc: TaskNotFoundException):
return JSONResponse(
status_code=exc.status_code,
content={"message": exc.detail},
)
Теперь наш API будет возвращать дружелюбное сообщение в случае ошибки.
Шаг 3: Логирование ошибок
Мы не можем игнорировать логи — они нужны для анализа и исправления проблем. Давайте настроим базовый логгер.
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled error: {str(exc)}")
return JSONResponse(
status_code=500,
content={"message": "Internal Server Error"},
)
Теперь при любой необработанной ошибке мы будем получать запись в логе.
Шаг 4: Middleware для обработки ошибок
Middleware обрабатывает запросы и ответы API на более глобальном уровне. Это позволяет перехватывать и модифицировать ошибки перед их отправкой пользователю.
from starlette.middleware.base import BaseHTTPMiddleware
class ErrorHandlerMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
try:
response = await call_next(request)
return response
except HTTPException as http_exc:
return JSONResponse({"message": http_exc.detail}, status_code=http_exc.status_code)
except Exception as exc:
logger.error(f"Unhandled error: {str(exc)}")
return JSONResponse({"message": "Something went wrong"}, status_code=500)
app.add_middleware(ErrorHandlerMiddleware)
С помощью этого middleware мы можем централизованно перехватывать ошибки.
Шаг 5: Обработка ошибок внешних API
Работа с внешними API сопряжена с дополнительными рисками, такими как тайм-ауты или сетевые сбои. Давайте добавим обработку таких ошибок.
import httpx
@app.get("/external-api")
async def call_external_api():
try:
async with httpx.AsyncClient() as client:
response = await client.get("https://some-external-api.com/data")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as http_exc:
logger.error(f"HTTP error occurred: {http_exc}")
raise HTTPException(status_code=400, detail="Failed to fetch data from external API")
except httpx.RequestError as req_exc:
logger.error(f"Request error: {req_exc}")
raise HTTPException(status_code=500, detail="External API request failed")
Здесь мы обрабатываем как ошибки HTTP-статуса, так и сетевые проблемы, предоставляя разные ответы пользователю.
Шаг 6: Настройка глобального обработчика
Глобальный обработчик ошибок позволяет централизованно перехватывать все исключения.
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}")
return JSONResponse(
status_code=500,
content={"message": "Internal Server Error. Please contact support."},
)
Теперь даже если ошибка не была обработана специфическими обработчиками, API вернёт консистентный ответ.
Шаг 7: Полная реализация
Теперь объединим всё вместе. Вот как может выглядеть финальная версия нашей системы обработки ошибок:
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import logging
import httpx
app = FastAPI()
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Пользовательские исключения
class TaskNotFoundException(HTTPException):
def __init__(self):
super().__init__(status_code=404, detail="Task not found")
# Глобальный обработчик ошибок
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}")
return JSONResponse(
status_code=500,
content={"message": "Internal Server Error. Please contact support."},
)
# Кастомный обработчик для TaskNotFoundException
@app.exception_handler(TaskNotFoundException)
async def task_not_found_handler(request: Request, exc: TaskNotFoundException):
return JSONResponse(
status_code=exc.status_code,
content={"message": exc.detail},
)
# Эндпоинт для тестирования ошибок
@app.get("/tasks/{task_id}")
async def get_task(task_id: int):
if task_id != 1: # Заглушка: считаем, что задача с id 1 существует
raise TaskNotFoundException()
return {"task_id": task_id, "name": "Learn FastAPI"}
# Взаимодействие с внешними API
@app.get("/external-api")
async def call_external_api():
try:
async with httpx.AsyncClient() as client:
response = await client.get("https://some-external-api.com/data")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as http_exc:
logger.error(f"HTTP error occurred: {http_exc}")
raise HTTPException(status_code=400, detail="Failed to fetch data from external API")
except httpx.RequestError as req_exc:
logger.error(f"Request error: {req_exc}")
raise HTTPException(status_code=500, detail="External API request failed")
Теперь наше приложение способно обрабатывать как стандартные HTTP-ошибки, так и пользовательские, логировать исключения и предотвращать сбои при взаимодействии с внешними API.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ