Перейти к содержанию

Работа с API на Python

Большинство внешних источников данных для дата-инженера — это REST API. Ты забираешь данные из CRM, биллинга, аналитических систем и складываешь в хранилище. В этой статье — практика работы с API: от простого GET-запроса до пагинации и retry.

requests — базовая библиотека

Python
import requests

response = requests.get("https://dummyjson.com/products/1", timeout=10)
response.raise_for_status()  # выбросит исключение при HTTP 4xx/5xx

product = response.json()
print(f"{product['title']}: ${product['price']}")
Text Only
Essence Mascara Lash Princess: $9.99

Всегда указывай timeout

Без timeout запрос может висеть бесконечно. 10–30 секунд — разумный default для API.

GET с параметрами

Python
response = requests.get(
    "https://dummyjson.com/products/search",
    params={"q": "phone", "limit": 5},
    timeout=10,
)
data = response.json()
print(f"Найдено: {data['total']}, получено: {len(data['products'])}")

POST — отправка данных

Python
response = requests.post(
    "https://dummyjson.com/products/add",
    json={"title": "Test Product", "price": 99},
    timeout=10,
)
response.raise_for_status()
print(response.json())

Авторизация

API-ключ в заголовке

Python
headers = {"Authorization": "Bearer YOUR_API_KEY"}

response = requests.get(
    "https://api.example.com/data",
    headers=headers,
    timeout=10,
)

Сессия — переиспользование соединений

Если делаешь много запросов к одному API — используй Session. Он переиспользует TCP-соединения и хранит общие заголовки:

Python
session = requests.Session()
session.headers.update({
    "Authorization": "Bearer YOUR_API_KEY",
    "Accept": "application/json",
})

# Все запросы через session используют те же заголовки и соединение
users = session.get("https://api.example.com/users", timeout=10).json()
orders = session.get("https://api.example.com/orders", timeout=10).json()

Session vs отдельные запросы

Session экономит время на установке TCP/TLS-соединений. Для 100+ запросов разница заметна.

Basic Auth

Python
response = requests.get(
    "https://api.example.com/data",
    auth=("username", "password"),
    timeout=10,
)

Пагинация

Большинство API отдают данные порциями. Два основных подхода:

Offset-based (skip/limit)

Python
import requests


def fetch_all_products(base_url: str) -> list[dict]:
    """Забирает все товары с пагинацией offset/limit."""
    all_products = []
    skip = 0
    limit = 30

    while True:
        response = requests.get(
            f"{base_url}/products",
            params={"skip": skip, "limit": limit},
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()

        products = data["products"]
        all_products.extend(products)

        if len(products) < limit:
            break  # последняя страница

        skip += limit

    return all_products


products = fetch_all_products("https://dummyjson.com")
print(f"Всего загружено: {len(products)}")

Cursor-based (next token)

Python
def fetch_all_items(base_url: str, api_key: str) -> list[dict]:
    """Забирает все элементы с cursor-пагинацией."""
    all_items = []
    cursor = None
    headers = {"Authorization": f"Bearer {api_key}"}

    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor

        response = requests.get(
            f"{base_url}/items",
            params=params,
            headers=headers,
            timeout=10,
        )
        response.raise_for_status()
        data = response.json()

        all_items.extend(data["items"])

        cursor = data.get("next_cursor")
        if not cursor:
            break

    return all_items

Offset vs Cursor

Offset-пагинация проста, но на больших датасетах может пропускать или дублировать строки (если данные меняются между запросами). Cursor-пагинация надёжнее — она привязана к конкретной позиции.

Retry и обработка ошибок

API может временно недоступен (5xx) или ограничивать частоту запросов (429). Для этого нужен retry.

Ручной retry

Python
import time
import requests


def get_with_retry(
    url: str,
    max_retries: int = 3,
    backoff: float = 1.0,
    **kwargs,
) -> requests.Response:
    """GET-запрос с retry и exponential backoff."""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10, **kwargs)

            if response.status_code == 429:
                # Rate limited — ждём и повторяем
                retry_after = int(response.headers.get("Retry-After", backoff))
                time.sleep(retry_after)
                continue

            response.raise_for_status()
            return response

        except requests.exceptions.RequestException:
            if attempt == max_retries - 1:
                raise
            time.sleep(backoff * (2 ** attempt))  # exponential backoff

    raise RuntimeError("Max retries exceeded")

tenacity — декларативный retry

Python
from tenacity import retry, stop_after_attempt, wait_exponential
import requests


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
def fetch_data(url: str) -> dict:
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

tenacity vs ручной retry

tenacity — стандарт для retry в Python. Декоратор @retry чище и гибче: можно задать условия retry, тип backoff, callback при ошибке. Установка: pip install tenacity.

JSON → DataFrame

Типичный паттерн DE: забрать данные из API и положить в таблицу.

Python
import pandas as pd
import requests


def load_products_to_df() -> pd.DataFrame:
    """Забирает товары из DummyJSON и возвращает DataFrame."""
    response = requests.get(
        "https://dummyjson.com/products",
        params={"limit": 10},
        timeout=10,
    )
    response.raise_for_status()

    products = response.json()["products"]

    df = pd.DataFrame(products)[["id", "title", "price", "category", "rating"]]
    return df


df = load_products_to_df()
print(df.head())
Text Only
   id                      title  price        category  rating
0   1  Essence Mascara Lash P...   9.99          beauty    4.94
1   2        Eyeshadow Palette   19.99          beauty    3.28
2   3  Powder Canister          14.99          beauty    3.82
3   4  Red Lipstick              12.99          beauty    4.94
4   5  Red Nail Polish            8.99          beauty    3.91

Вложенный JSON — json_normalize

Когда API возвращает вложенные объекты:

Python
import pandas as pd

data = [
    {"id": 1, "name": "Alice", "address": {"city": "Moscow", "zip": "101000"}},
    {"id": 2, "name": "Bob", "address": {"city": "Berlin", "zip": "10115"}},
]

df = pd.json_normalize(data)
print(df)
Text Only
   id   name address.city address.zip
0   1  Alice       Moscow      101000
1   2    Bob       Berlin       10115

Rate limiting

Многие API ограничивают количество запросов в секунду/минуту:

Python
import time
import requests

RATE_LIMIT = 5  # запросов в секунду
urls = [f"https://dummyjson.com/products/{i}" for i in range(1, 21)]

results = []
for url in urls:
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    results.append(response.json())
    time.sleep(1 / RATE_LIMIT)  # пауза между запросами

Не игнорируй rate limits

Если API вернул 429 (Too Many Requests) — это сигнал замедлиться. Игнорирование лимитов приведёт к бану API-ключа.

Что запомнить

Тема Ключевая мысль
timeout Всегда указывай timeout в запросах
Session Для 10+ запросов к одному API
Пагинация Offset для простых случаев, cursor для надёжности
Retry tenacity + exponential backoff
Результат pd.DataFrame(response.json()["items"]) — данные сразу в таблицу

Практика: попробуй забрать данные из DummyJSON — публичного API, который не требует авторизации.

Проверь себя

Источники