Представьте, что ваше приложение обращается к какому-нибудь внешнему сервису — например, чтобы получить свежие данные о погоде. Всё отлично, пока этот сервис работает стабильно. Но что, если он внезапно перестанет отвечать? Может быть, у них техработы, перегрузка или вообще DDoS-атака. Или запрос зависнет из-за плохого соединения. А может, сервер вернёт странный ответ — вместо ожидаемых данных просто пустой JSON. И не стоит забывать про лимиты: если превысили лимит запросов, ответ тоже может быть неожиданным.
Если такие ситуации не обрабатывать, приложение может просто развалиться на глазах у пользователя. А это не то, чего мы хотим. К тому же, умение грамотно обрабатывать подобные ошибки — это тот случай, когда вы выглядите профи и в боевом проекте, и на собеседовании.
Основные виды ошибок при работе с внешними API:
- Сетевые ошибки: проблемы с доступностью сервера, тайм-ауты, разрывы соединений.
- HTTP-ошибки: клиентские (например, 400 или 403) или серверные (500).
- Неверный формат данных: API возвращает не соответствующий ожиданиям ответ.
- Превышение лимитов: ограничения на количество запросов в минуту или в день.
- Ошибки авторизации: невалидный токен или ключ доступа.
Теперь давайте шаг за шагом разберем, как их обрабатывать.
Сетевые ошибки и тайм-ауты
Для работы с внешними 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). - Используем
HTTPExceptionFastAPI для возвращения информативных ответов на клиент.
Типичная ошибка: если забыть обрабатывать 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 не помешает вашему коду работать как часы.
ПЕРЕЙДИТЕ В ПОЛНУЮ ВЕРСИЮ