from __future__ import annotations import threading import time import unittest from dataclasses import dataclass from typing import Callable Clock = Callable[[], float] @dataclass class TokenBucket: """Thread-safe token-bucket rate limiter.""" capacity: float refill_rate: float clock: Clock = time.monotonic def __post_init__(self) -> None: if self.capacity <= 0: raise ValueError("capacity must be positive") if self.refill_rate <= 0: raise ValueError("refill_rate must be positive") self._tokens: float = self.capacity self._last_refill: float = self.clock() self._lock = threading.Lock() @property def tokens(self) -> float: with self._lock: self._refill() return self._tokens def consume(self, amount: float = 1.0) -> bool: """Consume tokens immediately, returning False if the bucket is short.""" if amount <= 0: raise ValueError("amount must be positive") if amount > self.capacity: raise ValueError("amount cannot exceed capacity") with self._lock: self._refill() if self._tokens < amount: return False self._tokens -= amount return True def time_until_available(self, amount: float = 1.0) -> float: """Return seconds until enough tokens exist for the requested amount.""" if amount <= 0: raise ValueError("amount must be positive") if amount > self.capacity: raise ValueError("amount cannot exceed capacity") with self._lock: self._refill() missing = amount - self._tokens return max(0.0, missing / self.refill_rate) def wait_and_consume(self, amount: float = 1.0) -> None: """Block until tokens are available, then consume them.""" while True: delay = self.time_until_available(amount) if delay == 0.0 and self.consume(amount): return time.sleep(delay) def _refill(self) -> None: now = self.clock() elapsed = max(0.0, now - self._last_refill) self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_rate) self._last_refill = now 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_consume_until_empty(self) -> None: clock = FakeClock() bucket = TokenBucket(capacity=2.0, refill_rate=1.0, clock=clock) self.assertTrue(bucket.consume()) self.assertTrue(bucket.consume()) self.assertFalse(bucket.consume()) def test_refills_over_time_without_exceeding_capacity(self) -> None: clock = FakeClock() bucket = TokenBucket(capacity=3.0, refill_rate=2.0, clock=clock) self.assertTrue(bucket.consume(3.0)) clock.advance(0.5) self.assertAlmostEqual(bucket.tokens, 1.0) clock.advance(10.0) self.assertAlmostEqual(bucket.tokens, 3.0) self.assertTrue(bucket.consume(3.0)) if __name__ == "__main__": unittest.main()