The Token Bucket algorithm is used in network switching and API rate limiting. Tokens are added to the bucket at a constant rate. In order to process an incoming packet or API request, a token must be removed from the bucket. If the bucket is empty, the request is dropped (or queued/throttled). This allows for sudden bursts of traffic if there are stored tokens.
import time
class TokenBucket:
def __init__(self, capacity: int, fill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.fill_rate = fill_rate
self.last_update = time.time()
def allow_request(self) -> bool:
# Refill tokens based on time elapsed
now = time.time()
elapsed = now - self.last_update
new_tokens = elapsed * self.fill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_update = now
# Check if we have enough tokens
if self.tokens >= 1:
self.tokens -= 1
return True
return Falseclass TokenBucket {
constructor(capacity, fillRatePerSecond) {
this.capacity = capacity;
this.tokens = capacity;
this.fillRate = fillRatePerSecond;
this.lastUpdate = Date.now();
}
allowRequest() {
const now = Date.now();
const elapsedSeconds = (now - this.lastUpdate) / 1000;
const newTokens = elapsedSeconds * this.fillRate;
this.tokens = Math.min(this.capacity, this.tokens + newTokens);
this.lastUpdate = now;
if (this.tokens >= 1) {
this.tokens -= 1;
return true;
}
return false;
}
}