Rate Limiting senza Redis: un Token Bucket Locale in Python

I tutorial sul rate limiting partono da Redis perché è ciò che usano i sistemi planetari. Ma un tool a processo singolo che chiama un'API di terze parti ha un problema molto più semplice: non superare N richieste per finestra. Puoi risolverlo in-process, gratis.

Il token bucket in venti righe

Un bucket contiene token che si ricaricano a ritmo fisso. Ogni richiesta consuma un token; quando il bucket è vuoto aspetti o rifiuti. I picchi vengono assorbiti fino alla capacità e il ritmo medio resta sotto controllo — esattamente il comportamento che le API si aspettano.

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate, self.capacity = rate, capacity
        self.tokens = capacity
        self.updated = time.monotonic()

    def allow(self) -> bool:
        now = time.monotonic()
        self.tokens = min(
            self.capacity,
            self.tokens + (now - self.updated) * self.rate,
        )
        self.updated = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

Persistere lo stato tra i riavvii

Un bucket in memoria dimentica tutto quando il processo riparte — ma le API contano il tuo utilizzo comunque. Salva i contatori su SQLite ogni pochi secondi e ricaricali all'avvio: il limiter sopravvive ai deploy senza servizi esterni.

def save(conn, key, tokens, updated):
    conn.execute(
        "INSERT OR REPLACE INTO buckets VALUES (?, ?, ?)",
        (key, tokens, updated),
    )
    conn.commit()

Insidie di concorrenza

Proteggi il bucket con un lock se lo condividono più thread, e ricorda che il limite è per processo: due repliche hanno ciascuna il proprio budget. Solo quando gestisci davvero una flotta ha senso passare a uno storage condiviso.

Semplice, locale, persistito. È tutto qui il pattern — ed è esattamente ciò che il nostro asset RateShield impacchetta con test e configurazione già pronti.