Асинхронная обработка запросов к 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 – это всего лишь начало!