from __future__ import annotations import threading import time import unittest from collections.abc import Callable class TokenBucket: """Thread-safe token-bucket rate limiter.""" def __init__( self, rate: float, capacity: float, *, clock: Callable[[], float] = time.monotonic, ) -> None: if rate <= 0: raise ValueError("rate must be positive") if capacity <= 0: raise ValueError("capacity must be positive") self.rate = rate self.capacity = capacity self._tokens = capacity self._clock = clock self._updated_at = clock() self._lock = threading.Lock() def _refill(self, now: float) -> None: elapsed = max(0.0, now - self._updated_at) self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) self._updated_at = now def consume(self, amount: float = 1.0) -> bool: """Consume tokens immediately, returning False when unavailable.""" if amount <= 0: raise ValueError("amount must be positive") if amount > self.capacity: return False with self._lock: self._refill(self._clock()) if self._tokens < amount: return False self._tokens -= amount return True def retry_after(self, amount: float = 1.0) -> float: """Return seconds until the requested tokens should be available.""" if amount <= 0: raise ValueError("amount must be positive") if amount > self.capacity: return float("inf") with self._lock: self._refill(self._clock()) return max(0.0, (amount - self._tokens) / self.rate) class TokenBucketTests(unittest.TestCase): def test_capacity_is_enforced(self) -> None: now = [0.0] bucket = TokenBucket(2.0, 2.0, clock=lambda: now[0]) self.assertTrue(bucket.consume(2.0)) self.assertFalse(bucket.consume()) def test_tokens_refill_over_time(self) -> None: now = [0.0] bucket = TokenBucket(2.0, 2.0, clock=lambda: now[0]) bucket.consume(2.0) now[0] = 0.5 self.assertEqual(bucket.retry_after(), 0.0) self.assertTrue(bucket.consume()) self.assertAlmostEqual(bucket.retry_after(), 0.5) if __name__ == "__main__": unittest.main()