Асинхронна обробка запитів до Google API з FastAPI

Ти, мабуть, вже знаєш, що Python підтримує асинхронне програмування за допомогою async і await.

Але чому це так важливо при роботі з зовнішніми API, такими як Google API?

Уяви, що ти відправляєш запит до Google Sheets, щоб отримати дані.

Якщо запит виконується синхронно, все, що ти можеш зробити — сидіти і чекати, поки Google поверне відповідь. Неефективно, правда?

Асинхронний підхід дозволяє твоєму застосунку продовжувати працювати, поки йде очікування відповіді від API.

Це особливо важливо, якщо твій застосунок обробляє велику кількість запитів. Використовуючи асинхронність,
ми можемо зменшити затримки, підвищити продуктивність і знизити навантаження на сервер.

---

Концепція асинхронної роботи з API

Асинхронна обробка задіює бібліотеку asyncio, вбудовану в Python, а для HTTP-запитів ми будемо використовувати httpx.

Ця бібліотека ідеально підходить для роботи з асинхронними запитами.

У поєднанні з FastAPI вона дозволяє легко інтегрувати зовнішні API, такі як Google API, з асинхронним підходом.

Головні переваги:

1. Паралельне виконання запитів. Ми можемо відправляти кілька запитів одночасно, не чекаючи завершення попереднього.
2. Оптимізація часу очікування. Поки сервер Google відповідає на один запит, ми можемо виконати інші задачі.
3. Покращення продуктивності. Ти можеш обробляти більше користувачів і запитів без додаткового навантаження.

---

Встановлення бібліотеки httpx

Перше, переконаймося, що httpx вже встановлено. Якщо ні, то... сюрприз! Встановлюємо:

pip install httpx

Створення базового асинхронного запиту

Робота з httpx дещо нагадує використання requests, але вона асинхронна! Ось базовий приклад простого GET-запиту:

import httpx
import asyncio

async def fetch_data():
    url = "https://jsonplaceholder.typicode.com/posts/1"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        print(response.json())  # Виводимо відповідь у форматі JSON

# Запускаємо асинхронну функцію
asyncio.run(fetch_data())

Тепер ти можеш легко адаптувати цей підхід до запитів Google API.

---

Інтеграція з Google API через асинхронні запити

Давай розберемо, як реалізувати асинхронний запит до Google Sheets API.

Ми будемо використовувати токен доступу, отриманий з попередньої лекції.

import httpx
from fastapi import FastAPI, Depends

app = FastAPI()

ACCESS_TOKEN = "your_google_api_access_token"

async def fetch_google_sheet(sheet_id, range_):
    url = f"https://sheets.googleapis.com/v4/spreadsheets/{sheet_id}/values/{range_}"
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers)
        response.raise_for_status()  # Перевіряємо наявність помилок
        return response.json()

@app.get("/sheet-data/{sheet_id}")
async def get_sheet_data(sheet_id: str, range_: str = "A1:D10"):
    data = await fetch_google_sheet(sheet_id, range_)
    return {"data": data}

Що тут відбувається:

1. Ми створили асинхронну функцію fetch_google_sheet, яка взаємодіє з Google Sheets API.
2. Ми використовуємо httpx.AsyncClient для виконання асинхронного запиту.
3. Ендпоінт /sheet-data/{sheet_id} дозволяє отримати дані з таблиці, вказавши її ID і діапазон клітинок.

Ти можеш протестувати цей API за допомогою curl або Postman:

curl -X GET "http://127.0.0.1:8000/sheet-data/your_sheet_id?range_=A1:B2"

---

Обробка великих обсягів даних

Коли ти працюєш з великими таблицями Google Sheets, одна з можливостей — розбити задачу
на кілька паралельних запитів. Ось приклад, як це зробити:

async def fetch_multiple_ranges(sheet_id, ranges):
    url = f"https://sheets.googleapis.com/v4/spreadsheets/{sheet_id}/values:batchGet"
    headers = {
        "Authorization": f"Bearer {ACCESS_TOKEN}"
    }
    params = {
        "ranges": ranges,
        "majorDimension": "ROWS"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers, params=params)
        response.raise_for_status()
        return response.json()

@app.get("/batch-sheet-data/{sheet_id}")
async def get_batch_sheet_data(sheet_id: str, ranges: str):
    ranges_list = ranges.split(",")  # Розділяємо діапазони, вказані через кому
    data = await fetch_multiple_ranges(sheet_id, ranges_list)
    return {"data": data}

Тепер ти можеш передавати кілька діапазонів клітинок:

curl -X GET "http://127.0.0.1:8000/batch-sheet-data/your_sheet_id?ranges=A1:B2,C1:D5"

Припускаю, це справді зручно, чи не так?

---

Оптимізація продуктивності

Коли взаємодієш із зовнішніми API, важливо пам'ятати про обмеження запитів (rate limiting). Ось кілька стратегій оптимізації:

• Використовуй "batch" запити (як ми зробили вище), щоб мінімізувати кількість звернень до API.
• Обробляй отримані дані локально, де це можливо.
• Обмежуй кількість одночасних запитів за допомогою бібліотеки asyncio.Semaphore.

Приклад обмеження запитів:

import asyncio

semaphore = asyncio.Semaphore(3)  # Максимум 3 одночасні запити

async def limited_fetch(url):
    async with semaphore:  # Очікування вільного слота
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            return response.json()

---

Тепер ти озброєний асинхронністю і готовий вирішувати задачі будь-якої складності. У реальних проєктах це відкриває
можливості для роботи з великими обсягами даних, інтеграції мікросервісів і побудови масштабованих рішень.

І так, Google API — це тільки початок!