Volání API v Pythonu: timeouty, retry a exponenciální backoff

Cizí API občas selže: vyprší spojení, server vrátí 503 nebo vás omezí kódem 429. Kód, který s tím nepočítá, buď spadne, nebo – hůř – visí donekonečna. Ukážeme si, jak psát klienta, který přechodné chyby ustojí a trvalé chyby rychle nahlásí.

1. Timeout vždy

import httpx

r = httpx.get("https://api.example.com/data")   # httpx: výchozí timeout 5 s

httpx má výchozí timeout 5 sekund. Knihovna requests výchozí timeout nemá – bez parametru timeout může čekat navždy. Nastavujte ho tedy explicitně a s rozmyslem:

timeout = httpx.Timeout(10.0, connect=3.0)   # 3 s na spojení, 10 s na ostatní
client = httpx.Client(timeout=timeout, base_url="https://api.example.com")

Rychlý timeout na navázání spojení odhalí nedostupný server, delší timeout čtení dá prostor pomalým odpovědím.

2. Které chyby opakovat

Situace	Opakovat?
Timeout, chyba spojení	ano
429 Too Many Requests	ano, podle `Retry-After`
500, 502, 503, 504	ano
400, 401, 403, 404, 422	ne – opakování nic nezmění

Chyby 4xx (kromě 429) znamenají chybu na vaší straně. Opakovat je je jen zbytečná zátěž.

3. Exponenciální backoff s jitterem

Mezi pokusy čekejte stále déle – 1 s, 2 s, 4 s, 8 s… – a přidejte náhodnou složku (jitter), aby se tisíc klientů po výpadku nevrátilo ve stejné vteřině.

import random
import time
import httpx

RETRY_STATUS = {429, 500, 502, 503, 504}

def get_s_retry(client: httpx.Client, url: str, max_pokusu: int = 5, **kw) -> httpx.Response:
    for pokus in range(1, max_pokusu + 1):
        try:
            r = client.get(url, **kw)
        except (httpx.TimeoutException, httpx.NetworkError):
            if pokus == max_pokusu:
                raise
            cekani = None
        else:
            if r.status_code not in RETRY_STATUS:
                r.raise_for_status()   # 4xx/5xx mimo seznam → výjimka hned
                return r
            if pokus == max_pokusu:
                r.raise_for_status()
            cekani = retry_after(r)

        if cekani is None:
            cekani = min(30, 2 ** (pokus - 1)) + random.uniform(0, 1)
        time.sleep(cekani)
    raise RuntimeError("nedosažitelné")

def retry_after(r: httpx.Response) -> float | None:
    hodnota = r.headers.get("Retry-After")
    if hodnota and hodnota.isdigit():
        return min(float(hodnota), 120)
    return None

Hlavička Retry-After může obsahovat i HTTP datum – pro jednoduchost ho tu ignorujeme a použijeme vlastní backoff. Horní limit čekání chrání před tím, aby vás server poslal spát na hodinu.

4. Totéž s knihovnou tenacity

pip install tenacity

import httpx
from tenacity import (
    retry, stop_after_attempt, wait_random_exponential,
    retry_if_exception, before_sleep_log,
)
import logging

log = logging.getLogger(__name__)

def je_docasna(exc: BaseException) -> bool:
    if isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)):
        return True
    if isinstance(exc, httpx.HTTPStatusError):
        return exc.response.status_code in {429, 500, 502, 503, 504}
    return False

@retry(
    retry=retry_if_exception(je_docasna),
    wait=wait_random_exponential(multiplier=1, max=30),
    stop=stop_after_attempt(5),
    before_sleep=before_sleep_log(log, logging.WARNING),
    reraise=True,
)
def nacti_uzivatele(client: httpx.Client, uid: int) -> dict:
    r = client.get(f"/users/{uid}")
    r.raise_for_status()
    return r.json()

reraise=True zajistí, že po posledním pokusu dostanete původní výjimku, ne obalovou RetryError. before_sleep_log zaloguje každý neúspěšný pokus – v produkci chcete vědět, že API zlobí, i když to nakonec prošlo.

Tenacity funguje i s async def funkcemi.

5. Vestavěné opakování spojení v httpx

transport = httpx.HTTPTransport(retries=3)
client = httpx.Client(transport=transport)

Pozor, tohle opakuje jen selhání při navazování spojení (ConnectError, ConnectTimeout), ne stavové kódy 5xx ani timeouty čtení. Je to doplněk, ne náhrada vlastní logiky.

6. Idempotence: co je bezpečné opakovat

GET, PUT a DELETE jsou podle HTTP idempotentní – opakování nezmění výsledek. POST ne: když vyprší timeout při vytváření objednávky, nevíte, jestli ji server zpracoval. Opakováním můžete vytvořit duplicitu.

Řešení: pokud API podporuje idempotency key (např. platební brány jako Stripe), pošlete jedinečný klíč v hlavičce a při opakování ten samý:

import uuid

klic = str(uuid.uuid4())
client.post("/objednavky", json=data, headers={"Idempotency-Key": klic})

Pokud API nic takového nemá, POST po timeoutu neopakujte slepě – nejdřív ověřte stav dotazem.

7. Stránkování

def vsechny_polozky(client: httpx.Client):
    url = "/items?limit=100"
    while url:
        r = get_s_retry(client, url)
        data = r.json()
        yield from data["items"]
        url = data.get("next")   # podle konkrétního API

Generátor vrací položky postupně, takže se nemusí načíst všechno do paměti.

8. Rate limiting na své straně

Když víte, že API povoluje např. 10 požadavků za sekundu, nečekejte na 429 – zpomalte sami. U synchronního kódu stačí time.sleep(0.1) mezi požadavky, u asynchronního asyncio.Semaphore pro omezení souběhu.

9. Circuit breaker v kostce

Pokud API selhává dlouhodobě, nemá smysl každý požadavek opakovat pětkrát. Vzor circuit breaker po N chybách po určitou dobu volání vůbec neprovádí a vrací chybu hned. Pro menší projekty stačí počítadlo chyb a čas posledního selhání; pro větší existují hotové knihovny.

Shrnutí

Timeout vždy, oddělený pro spojení a čtení.
Opakujte jen přechodné chyby: síť, 429, 5xx.
Exponenciální backoff s jitterem a horním limitem, respektujte Retry-After.
Pozor na opakování POST – používejte idempotency key.
Logujte každý neúspěšný pokus.