Обробка помилок і лімітів запитів до Google API

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

Розберімося, як зробити наше життя простішим і вберегти застосунок від загадкових "500 Internal Server Error".

Помилки — це як комарі на природі: непрошені, дратівливі, але з ними можна і треба боротися.

У контексті Google API помилки можуть виникати з різних причин:

• Неправильна конфігурація (наприклад, невірний API key або токен).
• Перевищення лімітів запитів (так, 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()

# Асинхронний endpoint для взаємодії з 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

Робота з токенами й оновлення квот

Якщо ви використовуєте service account або авторизацію через 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 можна врахувати за допомогою часових вікон.

---

Ефективна обробка помилок і керування лімітами — те, що перетворює "усе впало, ми виправимо через годину" на "ми зафіксували проблему і автоматично її вирішили".

Зусилля, вкладені в цю частину розробки, не лише заощадять нерви, але й зроблять застосунок стабільним і професійним.