Приклад обробки помилок при роботі з зовнішніми API

Уявіть, що ваш додаток звертається до якогось зовнішнього сервісу — наприклад, щоб отримати свіжі дані про погоду. Все ок, поки сервіс працює стабільно. Але що, якщо він раптово перестане відповідати? Можуть бути техроботи, перенавантаження або взагалі DDoS-атака. Або запит зависне через погане з'єднання. А може, сервер поверне дивну відповідь — замість очікуваних даних просто порожній JSON. І не забувай про ліміти: якщо перевищили ліміт запитів, відповідь теж може бути несподіваною.

Якщо такі ситуації не обробляти, додаток може просто розвалитися на очах у користувача. А це не те, чого ми хочемо. До того ж, вміння грамотно обробляти такі помилки — це той випадок, коли ти виглядаєш профі і в бойовому проєкті, і на співбесіді.

Основні види помилок при роботі з зовнішніми API:

1. Мережеві помилки: проблеми з доступністю сервера, тайм-аути, розриви з'єднань.
2. HTTP-помилки: клієнтські (наприклад, 400 або 403) або серверні (500).
3. Невірний формат даних: API повертає не той формат, що очікували.
4. Перевищення лімітів: обмеження на кількість запитів за хвилину або за день.
5. Помилки авторизації: невалідний токен або ключ доступу.

Тепер давайте крок за кроком розберемо, як їх обробляти.

Мережеві помилки і тайм-аути

Для роботи з зовнішніми API ми будемо використовувати бібліотеку httpx, яка підтримує асинхронні запити і зручна в користуванні.

Почнемо з обробки найпростішого сценарію: мережевої невдачі або тайм-ауту. Тайм-ауты дозволяють запобігти "вічному" очікуванню відповіді.

import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/weather")
async def get_weather():
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("https://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q=London")
            response.raise_for_status()  # Викинути помилку для статусів 4xx/5xx
            return response.json()
    except httpx.RequestError as exc:
        # Мережеві помилки (наприклад, недоступність сервера)
        raise HTTPException(status_code=503, detail=f"External API request failed: {exc}")
    except httpx.HTTPStatusError as exc:
        # Неправильний HTTP-статус (наприклад, 404 або 500)
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))

• Ми вказали timeout=5.0, щоб переривати запити, які займають більше 5 секунд.
• Метод raise_for_status автоматично викидає HTTPStatusError для помилок, пов'язаних з HTTP-статусами (4xx або 5xx).
• Використовуємо HTTPException FastAPI для повернення інформативних відповідей клієнту.

Типова помилка: якщо забути обробляти httpx.RequestError, твій сервер зависне на будь-якому підключенні до неіснуючого домену! Uptime твого сервісу може постраждати, тож не нехтуй часом очікування.

Обробка несподіваних даних

Припустимо, зовнішнє API відповідає JSON, але замість очікуваного об'єкта з погодою приходить несподіваний формат. Нам допоможе валідація даних через Pydantic.

from pydantic import BaseModel, ValidationError

class WeatherData(BaseModel):
    location: str
    temperature: float

@app.get("/weather")
async def get_weather():
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("https://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q=London")
            response.raise_for_status()
            data = response.json()
            # Валідація відповіді через Pydantic
            weather = WeatherData(
                location=data["location"]["name"],
                temperature=data["current"]["temp_c"]
            )
            return weather
    except KeyError as exc:
        # Якщо не знайдено потрібний ключ у відповіді
        raise HTTPException(status_code=502, detail="Malformed API response")
    except ValidationError as exc:
        # Якщо дані не валідні
        raise HTTPException(status_code=502, detail=f"Invalid data format: {exc}")

• Ми використовуємо Pydantic для перевірки формату даних відповіді.
• Обробляємо KeyError, якщо зовнішнє API раптово змінює структуру відповіді.

До речі, Pydantic такий швидкий, що його іноді використовують у проєктах, де швидкість критична (наприклад, у системах фінансових транзакцій).

Ліміти запитів і захист від "банів"

Деякі API мають обмеження на кількість запитів. Краще заздалегідь передбачити Retry-механізми або кешування відповідей.

Приклад роботи з retry:

from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(min=1, max=10), stop=stop_after_attempt(3))
async def fetch_data_with_retry(client, url):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

@app.get("/forecast")
async def get_forecast():
    async with httpx.AsyncClient() as client:
        try:
            data = await fetch_data_with_retry(client, "https://api.weatherapi.com/v1/forecast.json?key=YOUR_API_KEY&q=London")
            return data
        except Exception as exc:
            raise HTTPException(status_code=503, detail=f"Failed after retries: {exc}")

• Бібліотека tenacity автоматично повторює запити, якщо вони провалились.
• Використовуємо експоненціальну затримку між спробами, щоб уникнути навантаження на API.

Кастомні стратегії обробки помилок

У реальному житті простого "підняття виключення" недостатньо. Наприклад, можна повернути кешований результат у випадку тимчасової недоступності API.

import aiofiles
import json

CACHE_FILE = "weather_cache.json"

async def get_cached_weather():
    try:
        async with aiofiles.open(CACHE_FILE, mode="r") as file:
            return json.loads(await file.read())
    except FileNotFoundError:
        raise HTTPException(status_code=503, detail="No cached data available")

@app.get("/cached-weather")
async def get_weather():
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get("https://api.weatherapi.com/v1/current.json?key=YOUR_API_KEY&q=London")
            response.raise_for_status()
            # Зберігаємо вдалу відповідь у кеш
            async with aiofiles.open(CACHE_FILE, mode="w") as file:
                await file.write(json.dumps(response.json()))
            return response.json()
        except httpx.RequestError:
            # Повертаємо кешовані дані, якщо запит провалився
            return await get_cached_weather()

Підсумки

Тепер у тебе є все, щоб справлятися з типовими проблемами при роботі з зовнішніми API. Ти вмієш ловити тайм-аути, перевіряти, що прийшло у відповіді, і навіть додавати повто́ри з кешуванням, якщо щось пішло не так.

Такі штуки роблять додаток стійкішим, а користувачів — спокійнішими і задоволенішими. Бо ніхто не хоче бачити "помилка 503" замість прогнозу погоди. Використовуй ці знання і в реальних проєктах, і на співбесідах — нехай жодне зовнішнє API не завадить твоєму коду працювати як годинник.