Обработка ошибок при работе с внешними сервисами

Все мы знаем: программа работает идеально только на этапе демо для босса. В реальной жизни случаются таймауты, пропадает интернет-соединение, а внешние API возвращают ответы, которые вызывают слёзы даже у опытных разработчиков.

Ошибки при взаимодействии с внешними сервисами могут быть связаны с множеством факторов:

• Временные сбои сервера.
• Некорректные данные от клиента.
• Таймауты при подключении.
• Ошибки аутентификации (например, неверный API-ключ).
• Неверные запросы к API.
• Изменения в документации API без уведомления (да, такое бывает).

Обработка ошибок играет ключевую роль для обеспечения устойчивости вашего приложения. Пользователи не должны страдать из-за того, что внешний сервис не отвечает. Им нужны понятные и корректные сообщения, а вам — возможность зарегистрировать и исправить проблему.

Типичные ошибки при работе с внешними API

1. Проблемы с подключением

Возможно, сервер недоступен из-за сетевых проблем или API временно отключено на их стороне. Такие ошибки обычно проявляются как ConnectionError, TimeoutError или их аналоги в библиотеке httpx.

import httpx

async def get_data_from_api():
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get("https://example.com/nonexistent")
            response.raise_for_status()  # Проверка на HTTP-ошибки
            return response.json()
    except httpx.RequestError as exc:
        print(f"Ошибка во время подключения: {exc}")
        # Можно вернуть дефолтное значение или сообщение
        return {"error": "Сервер временно недоступен"}

# Вызов функции
data = await get_data_from_api()
print(data)

Если сервер "упал", обработка всё равно вернёт вам аккуратный ответ, вместо экрана с непонятной трассировкой.

2. Ошибки авторизации

Внешние API часто требуют аутентификацию, будь то через токены, ключи или OAuth. Если вы случайно отправите неверный токен или ключ, получите ошибку 401 Unauthorized или 403 Forbidden.

Обработаем эту ситуацию:

async def get_protected_data():
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.example.com/protected-resource",
                headers={"Authorization": "Bearer INVALID_TOKEN"}
            )
            response.raise_for_status()
    except httpx.HTTPStatusError as exc:
        if exc.response.status_code == 401:
            print("Ошибка авторизации. Проверьте ваш токен доступа!")
        elif exc.response.status_code == 403:
            print("Доступ запрещён. Вы не имеете прав для выполнения этого запроса.")
        else:
            print(f"HTTP ошибка: {exc}")
    return None

3. Неправильный формат данных

API может неожиданно вернуть вам XML вместо JSON (нет, это не шутка). Разбираемся с этим:

async def parse_response():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/weird-format")
        try:
            data = response.json()  # Предполагаем JSON
        except ValueError:
            print("Ошибка: Ответ от сервера не является корректным JSON.")
            return None
        return data

data = await parse_response()

Если API неожиданно решило "исправить" спецификацию, ваш код не упадёт, а аккуратно сообщит об ошибке.

Общая обработка исключений в httpx

Тут мы объединяем всё вместе. Главное правило: никогда не оставляйте потенциальные исключения необработанными!

async def fetch_data_with_error_handling():
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get("https://api.example.com/data")
            response.raise_for_status()  # Проверяем HTTP-ошибки
            return response.json()
    except httpx.TimeoutException:
        print("Ошибка: Запрос превысил установленные временные рамки.")
    except httpx.HTTPStatusError as exc:
        print(f"HTTP ошибка: Код {exc.response.status_code}, сообщение: {exc.response.text}")
    except httpx.RequestError as exc:
        print(f"Ошибка при запросе: {exc}")
    return None

Практические советы по обработке ошибок

1. Используйте try-except.
   Любая работа с внешним API должна быть окружена блоком try-except. Это ваш первый уровень защиты от нештатных ситуаций.
2. Логируйте ошибки.
   Вместо того, чтобы просто их печатать, используйте систему логирования, такую как logging. Это поможет в отладке и анализе проблем.

   
   import logging
   
   logging.basicConfig(level=logging.ERROR)
   
   try:
       async with httpx.AsyncClient() as client:
           response = await client.get("https://api.example.com")
   except httpx.RequestError as exc:
       logging.error(f"Ошибка: {exc}")
   

3. Повторные попытки.
   Используйте стратегию повторных попыток (retry) для временных ошибок. Например, с помощью цикла или библиотеки tenacity.

   
   from tenacity import retry, wait_fixed, stop_after_attempt
   
   @retry(wait=wait_fixed(2), stop=stop_after_attempt(3))
   async def fetch_with_retry():
       async with httpx.AsyncClient() as client:
           response = await client.get("https://api.example.com")
           response.raise_for_status()
           return response.json()
   

   Этот код пытается выполнить запрос три раза с интервалом в 2 секунды между попытками.

4. Показывайте информативные сообщения пользователям.
   Если ошибка связана с внешним API, опишите это пользователю понятным языком.

   
   async def user_friendly_api_call():
       try:
           async with httpx.AsyncClient() as client:
               response = await client.get("https://api.example.com/data")
               response.raise_for_status()
               return response.json()
       except httpx.RequestError:
           return {"error": "Внешний сервис временно недоступен. Попробуйте позже."}
   

Пример: Полноценная обработка ошибок для API

Давайте напишем пример, который объединяет всё, что мы прошли.

Допустим, мы интегрируемся с OpenWeatherMap API для получения погодных данных.

import httpx
from tenacity import retry, wait_fixed, stop_after_attempt
import logging

logging.basicConfig(level=logging.ERROR)

API_KEY = "ваш_api_ключ"

@retry(wait=wait_fixed(2), stop=stop_after_attempt(3))
async def fetch_weather(city: str):
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.openweathermap.org/data/2.5/weather",
                params={"q": city, "appid": API_KEY, "units": "metric"}
            )
            response.raise_for_status()
            return response.json()
    except httpx.HTTPStatusError as exc:
        logging.error(f"Ошибка API: Код {exc.response.status_code} - {exc.response.text}")
        return {"error": "Ошибка данных о погоде"}
    except httpx.RequestError as exc:
        logging.error(f"Ошибка при запросе: {exc}")
        return {"error": "Сетевая ошибка. Попробуйте позже."}

async def main():
    city = "Moscow"
    weather = await fetch_weather(city)
    print(weather)

# Вызов функции
# asyncio.run(main())

Этот пример выполняет запрос к API, повторяет его в случае временного сбоя и возвращает понятный ответ, если ошибка всё же произошла.

Таким образом, обработка ошибок при работе с внешними сервисами — это не просто "приятное дополнение",
а неотъемлемая часть качественного приложения.

1

Задача

Модуль 4: FastAPI, 16 уровень, 6 лекция

Недоступна

Простая обработка ошибок подключения

Простая обработка ошибок подключения

Открыть

1

Задача

Модуль 4: FastAPI, 16 уровень, 6 лекция

Недоступна

Обработка HTTP-ошибок при работе с API

Обработка HTTP-ошибок при работе с API

Открыть