hard Object-Oriented Programming

Token-Bucket Rate Limiter

The Brief

A fintech startup is launching a public API. Without rate limiting, one rogue client can crater the entire service — and the on-call engineer is you. The CTO has asked for a thread-safe, in-memory token-bucket rate limiter that the gateway can use to throttle requests per API key.

A token bucket has a capacity and a refill rate. Each request consumes tokens; when the bucket is empty, requests are denied. Tokens refill continuously over time, never exceeding capacity. The limiter must support multiple independent buckets keyed by string, and a decorator to wrap any function with rate limiting.

Acceptance Criteria

Tap a criterion to mark it off as you work. Tap a category header to expand or collapse.

1 Array

2 Array

3 Array

💡 Hints 0 / 5

Guest Mode Attempt this challenge freely, or log in to earn EXP & CodeCoins.

solution.py

Python 3

import time
import threading
from functools import wraps


class TokenBucket:
    def __init__(self, capacity, refill_rate):
        # TODO: validate, store capacity, refill_rate, tokens=capacity, last_refill
        # TODO: lock
        pass

    def _refill(self):
        # TODO: elapsed × rate, cap at capacity, advance last_refill
        pass

    def consume(self, tokens=1):
        # TODO: lock, refill, check, decrement, return
        pass


class RateLimiter:
    def __init__(self, capacity, refill_rate):
        # TODO: store config, init bucket dict, lock
        pass

    def allow(self, key, tokens=1):
        # TODO: lazy create bucket, consume
        pass


def rate_limited(limiter, key_fn):
    """Decorator factory. key_fn receives wrapped args and returns key str."""
    def decorator(func):
        # TODO: @wraps(func), check limiter.allow(key_fn(*args, **kwargs))
        pass
    return decorator

Console