Narcis Miclaus

10 min di lettura

Python, dalle fondamenta Lezione 39 / 60

Lavorare con le API: requests, retry, rate limit

La cassetta degli attrezzi HTTP in Python, i pattern di retry che non peggiorano le cose, e la gestione dei rate limit che ti tiene benvoluto.

L’altra metà dell’ingestion sono le API. Il servizio HTTP di qualcun altro ha i dati che ti servono. Scrivi un client Python, tiri pagine, scrivi su disco o su un database, lo rifai domani. Sembra facile. La prima versione naive funziona sempre. La seconda volta che la lanci, l’API ha una brutta giornata, e scopri che “facile” ha una lunga coda di modi di fallire: timeout, 503, 429, token scaduti, paginazione opaca, JSON che in realtà è XML. Questa lezione è la cassetta degli attrezzi per quella lunga coda.

Le librerie HTTP nel 2026

Tre librerie che incontrerai.

requests la classica. Solo sync. Non è cambiata molto negli anni; non ne ha bisogno. Il codice HTTP più semplice e leggibile che puoi scrivere in Python. Se stai scrivendo uno script una tantum e non ti importa della concorrenza, requests è ancora una scelta perfettamente valida nel 2026.

import requests

r = requests.get("https://api.example.com/orders", params={"limit": 100}, timeout=10)
r.raise_for_status()
data = r.json()

httpx moderna, sync e async, in larga parte drop-in compatibile con requests. Supporto HTTP/2 out of the box. La raccomandazione standard per il codice nuovo nel 2026, e quella che uso di default.

import httpx

with httpx.Client(timeout=10.0) as client:
    r = client.get("https://api.example.com/orders", params={"limit": 100})
    r.raise_for_status()
    data = r.json()

La versione async si legge quasi allo stesso modo:

import asyncio
import httpx

async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        responses = await asyncio.gather(*(client.get(u) for u in urls))
    return [r.json() for r in responses]

Quando hai 200 URL da colpire e una singola request impiega 200ms, la versione sync ci mette 40 secondi. La versione async ci mette 2. Quella differenza è il motivo per cui httpx esiste.

aiohttp solo async, l’opzione async più vecchia. Ancora ottima, ancora mantenuta, ancora comune in produzione. Se erediti una codebase che la usa, va benissimo. Per codice nuovo io scelgo httpx per l’unificazione sync/async.

Per questa lezione useremo httpx. I pattern si traducono in requests quasi riga per riga.

Le basi, fatte bene

Quattro cose da fare bene su ogni request:

r = client.get(
    "https://api.example.com/orders",
    params={"since": "2026-04-01", "limit": 100},
    headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    timeout=10.0,
)
r.raise_for_status()
data = r.json()
  • params parametri della query string come dict. Non concatenare stringhe negli URL; httpx codifica per te.
  • headers auth, accept, user-agent. Imposta uno User-Agent che identifichi la tua app; alcune API rifiutano quelli vuoti.
  • timeout sempre da impostare. Il default in alcune librerie è “aspetta per sempre”, che è un ottimo modo per appendere la tua pipeline dietro a una connessione bloccata. 10 secondi sono un punto di partenza ragionevole.
  • raise_for_status() solleva un’eccezione per le risposte 4xx/5xx. Senza di lui, il tuo codice continua allegramente con r.json() su una pagina di errore 500 e ottieni un confuso errore di parse JSON invece di un chiaro errore HTTP.

Un client che riusa la connessione è significativamente più veloce di chiamare httpx.get direttamente ogni volta, perché fa pooling delle connessioni TCP:

with httpx.Client(
    base_url="https://api.example.com",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
) as client:
    for endpoint in endpoints:
        r = client.get(endpoint)
        ...

Usa il client. Sempre. È una riga in più e uno speedup notevole su un’API chiacchierona.

Retry con tenacity

Alcuni fallimenti sono transitori. Il server ha avuto un singhiozzo. La connessione TCP è stata resettata. Un load balancer ha cambiato istanza. La risposta giusta è “aspetta un attimo e riprova”. La risposta sbagliata è “fallisci rumorosamente all’utente e metti un rerun manuale sul piatto del reperibile”.

La libreria è tenacity. Il decoratore è @retry, e compone qualche policy più piccola.

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
import httpx

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type((httpx.TransportError, httpx.HTTPStatusError)),
    reraise=True,
)
def fetch(client: httpx.Client, url: str) -> dict:
    r = client.get(url)
    r.raise_for_status()
    return r.json()

Cosa dice: prova fino a 5 volte; aspetta 1s, 2s, 4s, 8s tra i tentativi (con cap a 60s); ritenta su errori di rete ed errori HTTP; se falliamo ancora dopo 5 tentativi, rilancia l’ultima eccezione così il chiamante vede un errore vero.

La matematica dietro l’exponential backoff: ogni retry raddoppia l’attesa. La ragione non è superstizione, è che quando un servizio è sovraccarico, ritentare immediatamente non fa che aggiungere carico. L’exponential backoff lascia drenare la coda. Aggiungi un tocco di jitter (un wiggle casuale nel tempo di attesa) così mille client non fanno tutti retry nello stesso istante:

from tenacity import wait_random_exponential

wait=wait_random_exponential(multiplier=1, max=60),

Quello è “esponenziale fino a 60s, con jitter”. I docs di tenacity lo chiamano il default raccomandato per parlare con servizi esterni, ed è quello che cerco per primo.

Cosa ritentare, cosa non ritentare

Non ogni errore merita un retry. La divisione è grosso modo:

Ritentare: errori 5xx del server, 408 request timeout, 429 too many requests, errori di rete (ConnectError, ReadTimeout, RemoteProtocolError). Sono transitori.

Non ritentare: errori 4xx del client (eccetto 408/429). 400 bad request significa che la tua request era malformata, ritentare non cambia nulla. 401 significa che il tuo token non è valido, ritentare martella solo l’endpoint di auth. 404 significa che la risorsa non c’è. 422 significa che la validazione è fallita.

Tenacity ti dà la granularità:

def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, httpx.TransportError):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        status = exc.response.status_code
        return status >= 500 or status in (408, 429)
    return False

@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=60),
    retry=retry_if_exception(is_retryable),
    reraise=True,
)
def fetch(client: httpx.Client, url: str) -> dict:
    r = client.get(url)
    r.raise_for_status()
    return r.json()

Quello è il decoratore di retry production-grade. Rubalo.

Rate limit: rispetta la risposta

Quando un’API ti dice di rallentare, rallenta. La maggior parte delle API segnala questo con un HTTP 429 e un header Retry-After:

HTTP/1.1 429 Too Many Requests
Retry-After: 30

Onoralo:

def fetch(client: httpx.Client, url: str) -> dict:
    while True:
        r = client.get(url)
        if r.status_code == 429:
            wait_s = float(r.headers.get("Retry-After", "5"))
            log.warning("rate limited, sleeping %ss", wait_s)
            time.sleep(wait_s)
            continue
        r.raise_for_status()
        return r.json()

Quello è il pavimento: leggi l’header, dormi, ritenta. Il soffitto è il throttling proattivo: limita te stesso prima che debba farlo l’API. Se l’API permette 100 request al minuto, fai girare il tuo client a 90/minuto e non vedrai mai un 429.

La libreria limits ti dà un rate-limiter pulito:

from limits import RateLimitItemPerMinute
from limits.storage import MemoryStorage
from limits.strategies import MovingWindowRateLimiter

storage = MemoryStorage()
limiter = MovingWindowRateLimiter(storage)
quota = RateLimitItemPerMinute(90)

def fetch(client: httpx.Client, url: str) -> dict:
    while not limiter.hit(quota, "api.example.com"):
        time.sleep(0.1)
    r = client.get(url)
    r.raise_for_status()
    return r.json()

Oppure scriviti il tuo token bucket, sono circa 30 righe. Il punto è la gestione consapevole del rate. Le API ti throttlano perché qualcuno nel loro team ops si è preso una notifica alle 3 di notte per colpa di un client che si comportava male. Non essere quel client.

Dove possibile, batchizza. Se l’API ha un endpoint bulk (POST /orders/lookup che prende 100 ID), usalo invece di 100 GET singole.

Pattern di paginazione

Tre varianti che incontrerai, all’incirca in ordine crescente di gradevolezza.

Offset/limit ?offset=0&limit=100, poi ?offset=100&limit=100, eccetera. Il classico, ma ha un difetto: se vengono inserite righe mentre stai paginando, puoi perdere o duplicare righe. Accettabile per dataset statici, traballante per quelli vivi.

def paginate_offset(client: httpx.Client, url: str, limit: int = 100):
    offset = 0
    while True:
        r = client.get(url, params={"offset": offset, "limit": limit})
        r.raise_for_status()
        page = r.json()["data"]
        if not page:
            return
        yield from page
        offset += len(page)

Cursor-based l’API ritorna un token opaco next_cursor; glielo passi indietro per ottenere la pagina successiva. Stabile rispetto alle scritture, il default moderno.

def paginate_cursor(client: httpx.Client, url: str):
    cursor: str | None = None
    while True:
        params = {"cursor": cursor} if cursor else {}
        r = client.get(url, params=params)
        r.raise_for_status()
        body = r.json()
        yield from body["data"]
        cursor = body.get("next_cursor")
        if not cursor:
            return

Link header (RFC 5988) l’API mette l’URL della pagina successiva nell’header HTTP Link. GitHub usa questo. Meno comune ma elegante: non costruisci l’URL successivo, lo segui e basta.

def paginate_link(client: httpx.Client, url: str):
    while url:
        r = client.get(url)
        r.raise_for_status()
        yield from r.json()
        # httpx fa il parse degli header Link in r.links
        next_link = r.links.get("next")
        url = next_link["url"] if next_link else None

Leggi i docs, scegli quello giusto, scrivi un generator. I generator sono la forma ideale qui: il chiamante non deve sapere quante pagine ci sono né preoccuparsi dei confini di pagina.

Autenticazione

Le varianti:

  • API key negli header la più comune, e il posto giusto per loro. Authorization: Bearer <token> o un header custom tipo X-API-Key.
  • API key nelle query string alcune API legacy fanno così. Da evitare dove possibile: le query string finiscono nei log di accesso del server e nella cronologia del browser.
  • OAuth 2.0 client credentials per server-to-server. POST a un endpoint di token, ricevi un access token, lo usi per un’ora, lo refreshi.
  • OAuth 2.0 authorization code per i flussi “agisci per conto di un utente”. Redirect del browser, scope, refresh token. Non è quello che vuoi per i batch job.

Un piccolo wrapper per il refresh del token nel flusso client-credentials:

import time
from dataclasses import dataclass

@dataclass
class Token:
    value: str
    expires_at: float

class TokenManager:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self._token: Token | None = None

    def get(self, client: httpx.Client) -> str:
        if self._token and self._token.expires_at > time.time() + 60:
            return self._token.value
        r = client.post(self.token_url, data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        })
        r.raise_for_status()
        body = r.json()
        self._token = Token(body["access_token"], time.time() + body["expires_in"])
        return self._token.value

Il buffer di 60 secondi è voluto: refreshi leggermente prima della scadenza così non prendi un 401 a metà request perché l’orologio si è scostato.

Webhook vs polling

Inciso veloce. Se l’API offre webhook (“faremo POST al tuo URL quando succede qualcosa”), sono quasi sempre meglio del polling. Smetti di martellare un’API per “c’è qualcosa di nuovo?” dodici volte all’ora e lasci che la fonte pushi quando ci sono davvero novità. Il costo è far girare un piccolo server HTTP per riceverli. Per integrazioni ad alto volume o bassa latenza, questa è la via.

Il polling va bene per batch job a bassa frequenza, o quando non controlli l’infrastruttura per ricevere webhook, o quando l’API non li supporta.

Una nota sui client API generati da AI

Gli assistenti AI sono eccellenti nel produrre il boilerplate per questo tipo di codice. “Scrivimi un client httpx con retry tenacity che pagina per cursor e gestisce i rate limit” ti dà l’80% di un client funzionante in 15 secondi. I decoratori di retry, i loop di paginazione, i flussi di auth, tutto molto pattern-shaped, tutte cose che l’AI fa quasi alla perfezione.

Il trucco: gli assistenti AI a volte inventano nomi di endpoint che non esistono. Hanno visto diecimila client API e hanno fatto pattern-matching del tuo con uno simile, e ti producono con sicurezza codice che chiama GET /api/v2/users/me/orders quando l’API vera ha GET /v2/customer/orders. Verifica sempre i nomi degli endpoint, i nomi dei parametri e le forme delle response contro la documentazione vera dell’API prima di andare in produzione. Tratta l’output dell’AI come una bozza della struttura, non come la fonte di verità per l’API stessa.

Un esempio completo: pull paginato verso Parquet

Mettendo tutto insieme, uno script che tira ogni pagina di un’API paginata, ritenta sugli errori, rispetta i rate limit, e scrive su Parquet:

"""pull_orders.py — fetch all orders, write to Parquet."""
from __future__ import annotations
import logging
import time
from pathlib import Path

import httpx
import pyarrow as pa
import pyarrow.parquet as pq
from tenacity import (
    retry,
    retry_if_exception,
    stop_after_attempt,
    wait_random_exponential,
)

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("pull")

API = "https://api.example.com"
TOKEN = "..."


def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, httpx.TransportError):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code >= 500 or exc.response.status_code in (408, 429)
    return False


@retry(
    stop=stop_after_attempt(5),
    wait=wait_random_exponential(multiplier=1, max=60),
    retry=retry_if_exception(is_retryable),
    reraise=True,
)
def fetch(client: httpx.Client, url: str, params: dict | None = None) -> dict:
    r = client.get(url, params=params)
    if r.status_code == 429:
        wait_s = float(r.headers.get("Retry-After", "5"))
        log.warning("429 rate limited, sleeping %ss", wait_s)
        time.sleep(wait_s)
        r.raise_for_status()  # triggers retry
    r.raise_for_status()
    return r.json()


def paginate(client: httpx.Client, path: str):
    cursor: str | None = None
    while True:
        params = {"limit": 200, "cursor": cursor} if cursor else {"limit": 200}
        body = fetch(client, path, params=params)
        yield from body["data"]
        cursor = body.get("next_cursor")
        if not cursor:
            return


def main(out: Path) -> None:
    headers = {"Authorization": f"Bearer {TOKEN}", "User-Agent": "narcis-pull/1.0"}
    with httpx.Client(base_url=API, headers=headers, timeout=15.0) as client:
        rows = list(paginate(client, "/v1/orders"))
    log.info("fetched %d rows", len(rows))
    table = pa.Table.from_pylist(rows)
    pq.write_table(table, out, compression="zstd")
    log.info("wrote %s", out)


if __name__ == "__main__":
    main(Path("orders.parquet"))

Quella è tutta la forma: pagina pulitamente, ritenta sui fallimenti transitori, onora i rate limit, scrivi alla fine un file colonnare tipato. Buttalo sotto cron o sotto il tuo orchestrator, punta la pipeline di ingestion della lezione 38 sull’output Parquet, e hai cucito due metà di un ETL vero.

La lezione 41 è dove avvolgiamo un orchestrator attorno a tutto, ma quella è una storia per la prossima settimana.

Citations

Cerca