Уявіть свій 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.
ПЕРЕЙДІТЬ В ПОВНУ ВЕРСІЮ