token_bucket_rate_limiter

The oracle · bundled tests

from token_bucket_rate_limiter import TokenBucket

def test_caps():
    b = TokenBucket(rate=0, capacity=2)
    assert b.allow() and b.allow() and not b.allow()

The artifact

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        self.rate = rate; self.cap = capacity
        self.tokens = capacity; self.ts = time.monotonic()
    def allow(self, n: float = 1) -> bool:
        now = time.monotonic()
        self.tokens = min(self.cap, self.tokens + (now - self.ts) * self.rate); self.ts = now
        if self.tokens >= n:
            self.tokens -= n; return True
        return False