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, capacity: float, refill_rate: float, *, clock: Callable[[], float] = time.monotonic, ) -> None: if capacity <= 0: raise ValueError("capacity must be positive") if refill_rate <= 0: raise ValueError("refill_rate must be positive") self.capacity = float(capacity) self.refill_rate = float(refill_rate) self._tokens = self.capacity self._clock = clock self._updated_at = clock() self._lock = threading.Lock() def _refill(self) -> None: now = self._clock() elapsed = now - self._updated_at if elapsed > 0: self._tokens = min( self.capacity, self._tokens + elapsed * self.refill_rate, ) self._updated_at = now def consume(self, tokens: float = 1.0) -> bool: """Consume tokens immediately, returning False when unavailable.""" if tokens <= 0: raise ValueError("tokens must be positive") if tokens > self.capacity: return False with self._lock: self._refill() if self._tokens < tokens: return False self._tokens -= tokens return True def retry_after(self, tokens: float = 1.0) -> float: """Return seconds until the requested tokens should be available.""" if tokens <= 0: raise ValueError("tokens must be positive") if tokens > self.capacity: return float("inf") with self._lock: self._refill() return max(0.0, (tokens - self._tokens) / self.refill_rate) class FakeClock: def __init__(self) -> None: self.now = 0.0 def __call__(self) -> float: return self.now def advance(self, seconds: float) -> None: self.now += seconds class TokenBucketTests(unittest.TestCase): def test_consumption_and_refill(self) -> None: clock = FakeClock() bucket = TokenBucket(2, 1, clock=clock) self.assertTrue(bucket.consume(2)) self.assertFalse(bucket.consume()) clock.advance(1) self.assertTrue(bucket.consume()) def test_retry_after(self) -> None: clock = FakeClock() bucket = TokenBucket(5, 2, clock=clock) self.assertTrue(bucket.consume(5)) self.assertEqual(bucket.retry_after(3), 1.5) clock.advance(1) self.assertEqual(bucket.retry_after(3), 0.5) if __name__ == "__main__": unittest.main()