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._clock = clock self._tokens = capacity 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, tokens: float = 1.0) -> bool: """Return True and spend tokens when enough are available.""" if tokens <= 0: raise ValueError("tokens must be positive") with self._lock: now = self._clock() self._refill(now) if self._tokens < tokens: return False self._tokens -= tokens return True def allow(self) -> bool: """Convenience method for one-token requests.""" return self.consume(1.0) def time_until_available(self, tokens: float = 1.0) -> float: """Return seconds until the requested number of tokens is available.""" if tokens <= 0: raise ValueError("tokens must be positive") if tokens > self.capacity: raise ValueError("tokens cannot exceed bucket capacity") with self._lock: now = self._clock() self._refill(now) if self._tokens >= tokens: return 0.0 return (tokens - self._tokens) / self.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_burst_then_limit(self) -> None: clock = FakeClock() bucket = TokenBucket(rate=2.0, capacity=3.0, clock=clock) self.assertTrue(bucket.allow()) self.assertTrue(bucket.allow()) self.assertTrue(bucket.allow()) self.assertFalse(bucket.allow()) def test_refills_over_time(self) -> None: clock = FakeClock() bucket = TokenBucket(rate=2.0, capacity=3.0, clock=clock) self.assertTrue(bucket.consume(3.0)) self.assertFalse(bucket.allow()) clock.advance(0.5) self.assertTrue(bucket.allow()) self.assertFalse(bucket.allow()) clock.advance(1.0) self.assertTrue(bucket.consume(2.0)) if __name__ == "__main__": unittest.main()