Использование библиотеки 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)