Використання бібліотеки httpx для роботи з зовнішніми API

Коли йдеться про роботу з зовнішніми API, ваш бекенд найчастіше виконує роль «посильного». Ви відправляєте запити, отримуєте відповіді й передаєте інформацію далі. Здається, що тут нічого складного: «взяв і відправив запит». Проте в сучасному світі асинхронності ефективне використання ресурсів сервера стає критично важливим, і саме тут у гру вступає httpx.

Переваги використання httpx

1. Асинхронність: на відміну від requests, httpx підтримує асинхронні запити «із коробки». Це дозволяє вашому додатку не «зависати» під час очікування відповіді від зовнішнього сервісу.
2. Сучасний API: httpx надає потужні інструменти, такі як управління сесіями (Session Management), підключення через проксі та таймаути.
3. Сумісність: httpx працює як з синхронними, так і з асинхронними запитами, надаючи гнучкість у розробці.
4. Підтримка HTTP/2: якщо ваш API-сервіс підтримує HTTP/2, то httpx готовий до цього прямо «із коробки».

---

Встановлення httpx

Спочатку додамо httpx у наш проєкт через pip. Відкрийте термінал і виконайте:

pip install httpx

Не забудьте оновити requirements.txt у проєкті, щоб усі залежності були зафіксовані (в уроках з Docker ми вже додавали цей файл):

pip freeze > requirements.txt

---

Основні функції httpx

Переходимо до практики. Почнемо з простого: відправлення синхронних запитів. Потім перейдемо до асинхронних запитів, які стануть у нагоді нам у FastAPI.

Синхронні запити

import httpx

url = "https://jsonplaceholder.typicode.com/posts"
response = httpx.get(url)

# Перевіряємо статус відповіді
if response.status_code == 200:
    # Отримуємо дані у форматі JSON
    data = response.json()
    print(data)
else:
    print(f"Помилка: {response.status_code}")

Тут ми використовуємо метод httpx.get() для відправлення GET-запиту до тестового API. Якщо статус відповіді 200 (OK), виводимо дані.

Асинхронні запити

Асинхронні запити — це як улюблений спосіб програмістів сказати серверу: «Я чекаю, але при цьому не сумую». З FastAPI ви часто будете працювати саме так.

import httpx
import asyncio

async def fetch_posts():
    url = "https://jsonplaceholder.typicode.com/posts"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Помилка: {response.status_code}")

# Запуск асинхронної функції
asyncio.run(fetch_posts())

Зверніть увагу на використання httpx.AsyncClient — це клієнт, який дозволяє відправляти асинхронні запити. Ми також використовуємо async with для коректного управління з'єднанням.

---

Реальний приклад з FastAPI

Але достатньо абстракцій — давайте вбудуємо це в наш додаток FastAPI.

Створення ендпоінта для отримання даних із зовнішнього API

1. Створимо новий файл external_api.py:

from fastapi import APIRouter, HTTPException
import httpx

router = APIRouter()

@router.get("/external-data")
async def get_external_data():
    url = "https://jsonplaceholder.typicode.com/posts"
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
    
    # Якщо відповідь невдала, повертаємо помилку
    if response.status_code != 200:
        raise HTTPException(status_code=500, detail="Помилка при роботі з зовнішнім API")

    return response.json()

2. Підключимо цей роутер у головний файл додатку main.py:

from fastapi import FastAPI
from external_api import router as external_api_router

app = FastAPI()

app.include_router(external_api_router, prefix="/api")

Тепер запит до вашого локального сервера (наприклад, http://127.0.0.1:8000/api/external-data) буде автоматично проксувати дані з зовнішнього API.

---

Обробка помилок

Працюючи з зовнішніми API, помилки трапляються часто — від тимчасової недоступності сервісу до неправильних даних. Ось як їх можна обробляти за допомогою httpx.

async with httpx.AsyncClient() as client:
    try:
        response = await client.get(url, timeout=10)
        response.raise_for_status()  # Генерує виняток, якщо статус-код не 2xx
    except httpx.HTTPStatusError as exc:
        print(f"HTTP помилка: {exc.response.status_code}")
    except httpx.RequestError as exc:
        print(f"Помилка запиту: {exc}")

Ми використовуємо метод raise_for_status для перевірки успішності запиту. Також обробляємо можливі мережеві помилки.

---

Таймаути, заголовки та параметри

Бібліотека httpx надає гнучкі налаштування для запитів, такі як додавання заголовків, параметрів і керування таймаутами.

Щоб запобігти «вічному очікуванню» відповіді від сервера, вкажемо таймаут:

async with httpx.AsyncClient(timeout=5.0) as client:
    response = await client.get(url)

Якщо потрібно передати заголовки (наприклад, API-ключ), це робиться так:

headers = {"Authorization": "Bearer YOUR_TOKEN"}
async with httpx.AsyncClient(headers=headers) as client:
    response = await client.get(url)

Передача query-параметрів через httpx:

params = {"userId": 1}
async with httpx.AsyncClient() as client:
    response = await client.get(url, params=params)