Token bucket algorithm in Rest API

Introduction

The token bucket algorithm helps control how many requests a user can make to a server in a certain time. It stops too many requests from overloading the system.

To limit how many API calls a user can make per minute.

To prevent a website from slowing down due to too many visitors at once.

To control data flow in a network so it stays smooth and steady.

To allow short bursts of activity but keep the overall rate steady.

Syntax

Rest API

import time

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate  # tokens per second
        self.last_refill_timestamp = time.time()

    def allow_request(self, tokens=1):
        now = time.time()
        elapsed = now - self.last_refill_timestamp
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill_timestamp = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        else:
            return False

The capacity is the max tokens the bucket can hold.

The refill_rate controls how fast tokens are added back.

Examples

This example creates a bucket that holds 5 tokens and refills 1 token per second. It checks if a request can be allowed.

Rest API

bucket = TokenBucket(capacity=5, refill_rate=1)
if bucket.allow_request():
    print("Request allowed")
else:
    print("Request denied")

This tries 12 requests quickly. Only the first 10 will be allowed because of the capacity.

Rest API

bucket = TokenBucket(capacity=10, refill_rate=2)
for _ in range(12):
    if bucket.allow_request():
        print("Allowed")
    else:
        print("Denied")

Sample Program

This program creates a token bucket with 3 tokens max and refills 1 token per second. It tries 5 requests, waiting half a second between each. You will see some requests denied because tokens run out.

Rest API

import time

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate  # tokens per second
        self.last_refill_timestamp = time.time()

    def allow_request(self, tokens=1):
        now = time.time()
        elapsed = now - self.last_refill_timestamp
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill_timestamp = now

        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        else:
            return False

bucket = TokenBucket(capacity=3, refill_rate=1)

for i in range(5):
    if bucket.allow_request():
        print(f"Request {i+1}: Allowed")
    else:
        print(f"Request {i+1}: Denied")
    time.sleep(0.5)

Important Notes

Tokens refill over time, so waiting lets more requests pass.

Requests use tokens; if none left, requests are denied.

You can adjust capacity and refill rate to control traffic.

Summary

The token bucket algorithm controls request rates by using tokens.

Tokens refill steadily, allowing bursts but limiting overall rate.

This helps keep servers safe from too many requests at once.