Обработка ошибок и лимитов запросов к Google API

Сегодня у нас немного сложная, но очень важная тема — обработка ошибок и лимитов запросов к Google API.

Давайте разберёмся, как сделать нашу жизнь проще и уберечь приложение от загадочных "500 Internal Server Error".

Ошибки — это как комары на природе: непрошенные, назойливые, но с ними можно бороться.

В контексте Google API ошибки могут возникнуть по разным причинам:

• Неправильная конфигурация (например, неверный ключ API или токен).
• Превышение лимитов запросов (да, Google — не бесконечно терпеливый робот).
• Ошибки сети (интернет не всегда стабилен, особенно, если ваш сервер застрял на Wi-Fi в кафе).
• Неверные данные (например, подаете на вход невалидный JSON).

Без грамотной обработки ошибок можно случайно вызвать каскадный эффект: одно упавшее действие тянет за собой всё приложение.

Причем, если фронтенд подаст пользователю ошибку "Не удалось обработать запрос", это не всегда вдохновляет его снова посетить вашу платформу.

---

Основные типы ошибок Google API

Google API возвращает ошибки в формате JSON. Каждая ошибка содержит код состояния (HTTP Status Code) и дополнительную информацию.

Вот типичные ошибочные коды:

Все эти коды помогают понять, что именно пошло не так. Теперь давайте научимся их обрабатывать.

---

Обработка ошибок в FastAPI

Работа с Google API — это всегда HTTP-запросы. В библиотеке httpx, которую мы использовали ранее, можно обрабатывать ошибки через блоки try и except. Вот пример:

import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()

# Асинхронный эндпоинт для взаимодействия с Google API
@app.get("/google-data")
async def get_google_data():
    try:
        # Пример запроса к Google API
        async with httpx.AsyncClient() as client:
            response = await client.get("https://www.googleapis.com/sheets/v4/spreadsheets/some-id")
            response.raise_for_status()  # Выброс ошибки для HTTP-кодов 4xx/5xx

        # Парсим результат
        data = response.json()
        return {"data": data}

    except httpx.HTTPStatusError as e:
        # Обработка ошибок HTTP-запросов
        if e.response.status_code == 401:
            raise HTTPException(status_code=401, detail="Неправильный токен аутентификации.")
        elif e.response.status_code == 403:
            raise HTTPException(status_code=403, detail="Доступ запрещён. Проверьте права доступа.")
        elif e.response.status_code == 429:
            raise HTTPException(status_code=429, detail="Превышен лимит запросов. Подождите немного.")
        else:
            raise HTTPException(status_code=500, detail="Произошла неизвестная ошибка.")

    except Exception as e:
        # Очистка любых других необработанных ошибок
        raise HTTPException(status_code=500, detail="Что-то пошло не так: " + str(e))

Что здесь происходит?

1. Обрабатываем ошибки HTTP-запросов: response.raise_for_status() выбрасывает исключение, если статус-код 4xx или 5xx.
2. Диагностируем код ошибки: Мы проверяем код состояния и выдаем соответствующую информацию пользователю.
3. Ловим неописанные проблемы: Любые другие ошибки обрабатываются в общем блоке except.

---

Лимиты запросов: что и как?

Google API позволяет отправлять ограниченное количество запросов в рамках заданного времени.

Это называется квотой запросов. Если вы превышаете лимиты, сервер вернет статус-код 429 (Too Many Requests) с сообщением вроде:

{
  "error": {
    "code": 429,
    "message": "Quota exceeded for quota metric.",
    "errors": [
      {
        "reason": "quotaExceeded",
        "message": "Quota exceeded for quota metric."
      }
    ]
  }
}

Как избежать превышения лимитов?

1. Планируйте квоты: в Google Cloud Console можно проверить и увеличить лимиты. Например, для Sheets API квота составляет 500 запросов в минуту.
2. Кэшируйте запросы: если вы часто запрашиваете одни и те же данные, сохраните их локально и обновляйте только при необходимости.
3. Используйте задержки (throttling): это особенно полезно, если вы обрабатываете данные в цикле.

Вот пример с использованием задержек:

import asyncio
import httpx

async def fetch_google_data(urls):
    results = []
    async with httpx.AsyncClient() as client:
        for url in urls:
            try:
                response = await client.get(url)
                response.raise_for_status()
                results.append(response.json())
            except httpx.HTTPStatusError as e:
                print(f"Ошибка при запросе {url}: {e}")
            await asyncio.sleep(0.1)  # Пауза между запросами
    return results

Работа с токенами и обновлением квот

Если вы используете сервисный аккаунт или авторизацию через OAuth2, Google автоматически разрешит вам определённое количество запросов.

Токены доступа истекают, поэтому важно их обновлять. Этот процесс уже разбирался в предыдущих лекциях.

---

Логирование и мониторинг ошибок

Логирование — это первый шаг к пониманию проблем. Используйте библиотеку, например, logging, чтобы сохранять ошибки:

import logging
from fastapi import FastAPI

logging.basicConfig(level=logging.INFO)

app = FastAPI()

@app.get("/log-error")
async def log_error_example():
    try:
        # Здесь может быть ваш код для работы с Google API
        raise ValueError("Пример ошибки для демонстрации")
    except Exception as e:
        logging.error(f"Произошла ошибка: {e}")
        return {"message": "Ошибка зафиксирована"}

Вы можете хранить логи локально или отправлять их в облачные сервисы, такие как Sentry или Prometheus, для более глубокой аналитики.

---

Пример интеграции: умное ограничение запросов

Давайте реализуем пример системы, которая динамически управляет запросами с учётом лимитов:

import asyncio
import httpx

class GoogleAPIClient:
    def __init__(self, limit_per_minute: int):
        self.limit_per_minute = limit_per_minute
        self.requests_made = 0

    async def make_request(self, url: str):
        if self.requests_made >= self.limit_per_minute:
            await asyncio.sleep(60)  # Ждем минуту
            self.requests_made = 0

        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            response.raise_for_status()
            self.requests_made += 1
            return response.json()

# Использование клиента
client = GoogleAPIClient(limit_per_minute=10)

@app.get("/throttled-api")
async def throttled_api():
    url = "https://www.googleapis.com/sheets/v4/spreadsheets/sheet-id"
    data = await client.make_request(url)
    return {"data": data}

Этот пример показывает, как ограничения API можно обойти с помощью временных окон.

---

Эффективная обработка ошибок и управление лимитами — это то, что превращает "всё упало, мы исправим через час" в "мы зафиксировали проблему и автоматически её решили".

Усилия, вложенные в эту часть разработки, не только сэкономят вам нервы, но и сделают приложение стабильным и профессиональным.