1 The Problem

We want a rate limiter: a component that allows, say, 5 requests per minute per user and rejects the rest. We will build the token bucket algorithm: each user has a bucket that refills over time, and each request spends a token. It teaches controlling flow over time — essential for protecting any service.

Where this shows up: every public API (to prevent abuse), login attempt throttling, DDoS protection, fair-use quotas, traffic shaping. Rate limiting is how services protect themselves from being overwhelmed.

2 How to Think About It

Think about the token bucket, before any code:

The plan — in plain English

1. Each user has a bucket holding up to N tokens. → 2. The bucket refills over time — so many tokens per second, up to the maximum. → 3. Each request tries to spend one token: if the bucket has one, allow it and spend it; if empty, reject. → This naturally allows bursts up to N, then slows to the refill rate.

3 The Build — explained part by part

Here is a complete token-bucket rate limiter. The refill maths is testable with a fake clock. Each part is explained below.

Pythonratelimit.py

import time

class RateLimiter:
    def __init__(self, capacity, refill_per_second):
        self.capacity = capacity                # max tokens
        self.refill_rate = refill_per_second    # tokens added per second
        self.buckets = {}                       # user -> (tokens, last_time)

    def allow(self, user, now=None):
        now = now if now is not None else time.time()
        tokens, last = self.buckets.get(user, (self.capacity, now))

        # Refill: add tokens for the time that has passed, up to capacity.
        elapsed = now - last
        tokens = min(self.capacity, tokens + elapsed * self.refill_rate)

        if tokens >= 1:
            self.buckets[user] = (tokens - 1, now)   # spend one
            return True
        else:
            self.buckets[user] = (tokens, now)       # none to spend
            return False

if __name__ == &quot;__main__&quot;:
    # 3 requests allowed, refilling 1 per second.
    limiter = RateLimiter(capacity=3, refill_per_second=1)
    for i in range(5):
        print(f&quot;Request {i+1}: {&#x27;allowed&#x27; if limiter.allow(&#x27;alice&#x27;) else &#x27;REJECTED&#x27;}&quot;)

What each part does — in plain words

capacity and refill_rate — the two settings: how many tokens the bucket holds, and how fast it refills. Together they define “N per time window”.

self.buckets — each user’s state: how many tokens they have and when we last updated. New users start with a full bucket.

now=None ... time.time() — we let the caller pass the current time. In real use it defaults to now; in tests we pass a fake time, so we can test without waiting.

tokens = min(capacity, tokens + elapsed * refill_rate) — the refill: add tokens for the seconds that have passed, but never exceed capacity. This is the heart of the algorithm.

if tokens >= 1: — if there is a token, spend it (subtract 1) and allow the request; otherwise reject. Allowing bursts up to capacity, then throttling to the refill rate, falls out naturally.

Common mistakes — and how to avoid them

✗ Hard-coding time.time() inside the logic — then you cannot test without waiting.

✓ Pass now as a parameter so tests control the clock.

✗ Forgetting to cap at capacity — tokens grow forever and limiting never kicks in.

✓ Use min(capacity, ...) when refilling.

✗ Resetting the bucket fully instead of refilling gradually — loses the smooth behaviour.

✓ Add tokens proportional to elapsed time, not all at once.

4 Test & Prove Each Part

We test the limiter with a fake clock — passing explicit times — so we can test refill behaviour instantly without real waiting.

Requests within capacity are allowed

Requests beyond capacity are rejected

Tokens refill as time passes

Pythontest_ratelimit.py

class RateLimiter:
    def __init__(self, capacity, refill_per_second):
        self.capacity = capacity
        self.refill_rate = refill_per_second
        self.buckets = {}

    def allow(self, user, now):
        tokens, last = self.buckets.get(user, (self.capacity, now))
        elapsed = now - last
        tokens = min(self.capacity, tokens + elapsed * self.refill_rate)
        if tokens >= 1:
            self.buckets[user] = (tokens - 1, now)
            return True
        self.buckets[user] = (tokens, now)
        return False


def test_allows_within_capacity():
    &quot;&quot;&quot;The first 3 requests (capacity 3) are allowed.&quot;&quot;&quot;
    limiter = RateLimiter(3, 1)
    assert limiter.allow(&quot;u&quot;, now=0) is True
    assert limiter.allow(&quot;u&quot;, now=0) is True
    assert limiter.allow(&quot;u&quot;, now=0) is True


def test_rejects_over_capacity():
    &quot;&quot;&quot;The 4th request at the same instant is rejected.&quot;&quot;&quot;
    limiter = RateLimiter(3, 1)
    for _ in range(3):
        limiter.allow(&quot;u&quot;, now=0)
    assert limiter.allow(&quot;u&quot;, now=0) is False


def test_refills_over_time():
    &quot;&quot;&quot;After enough time, a token is available again.&quot;&quot;&quot;
    limiter = RateLimiter(3, 1)
    for _ in range(3):
        limiter.allow(&quot;u&quot;, now=0)
    assert limiter.allow(&quot;u&quot;, now=0) is False      # empty
    assert limiter.allow(&quot;u&quot;, now=2) is True       # 2s later, refilled

Run with pytest -v. Passing an explicit now turns time into something we control, so the refill test proves tokens come back after 2 seconds — without the test sitting for 2 real seconds. Controlling the clock is the key to testing time-based logic.

5 The Interface

allowlimiter.allow(user)true if permitted

Behaviour (capacity 3, refill 1/s)

Requests 1-3: allowed (burst)
Request 4: REJECTED
After 1 second: 1 more allowed

6 Run It & Automate It

Run it locally

python3 ratelimit.py
See 3 requests allowed then rejected. Add time.sleep(1) between calls to watch tokens refill.

Jenkins runs the tests automatically — each line explained below.

What you should see when it works

Terminala real run

Request 1: allowed
Request 2: allowed
Request 3: allowed
Request 4: REJECTED
Request 5: REJECTED

If it breaks — how to fix it

🚨 Nothing is ever rejected.

Your refill is too generous or not capped. Check min(capacity, ...) and the rate.

🚨 Everything is rejected.

New users should start with a full bucket. Check the default in .get(user, (capacity, now)).

GroovyJenkinsfile

// Jenkinsfile — automated tests on every change.
pipeline {
    agent any
    stages {
        stage(&#x27;Get the code&#x27;) { steps { checkout scm } }       // download the code
        stage(&#x27;Install&#x27;) { steps { sh &#x27;python3 -m venv venv && . venv/bin/activate && pip install pytest&#x27; } }  // test tool
        stage(&#x27;Test&#x27;) { steps { sh &#x27;. venv/bin/activate && pytest -v&#x27; } }   // run every test
    }
    post {
        success { echo &#x27;All tests passed.&#x27; }
        failure { echo &#x27;A test failed — look above.&#x27; }
    }
}

🎯 Try this next — make it yours

Per-endpoint limits. Different limits for different actions. (Teaches: multiple buckets.)
Sliding window. Implement the alternative sliding-log algorithm. (Teaches: another approach.)
Wire to the API. Add it to the REST API project to throttle requests. (Teaches: integration.)

What you learned

You built a token-bucket rate limiter — the algorithm protecting nearly every API — and learned to test time-based logic with a controllable clock. Controlling flow over time is essential systems knowledge. Related: time, Object-Oriented Python.

Rate Limiter