The Throttle Wall

A hard Python interview practice problem on DataDriven. Write the Python and run it against test cases, with instant feedback.

Domain
Python
Difficulty
hard
Seniority
L4
Asked in
Google

Problem

An API gateway throttles each client independently: within any window_ms window a client may keep at most max_requests accepted calls. The operations list opens with ['init', max_requests, window_ms] to configure the limiter and is followed by ['is_allowed', client_id, timestamp] decisions; return one result per entry, None for the configure step and then True or False for each decision. A call is allowed only when the client has fewer than max_requests accepted calls still inside the half-open window (timestamp - window_ms, timestamp], and a denied call is never recorded, so a throttled client becomes eligible again as soon as its earlier accepted calls age out.

Stop the abusers. Let the rest through.

Example

Input:

operations = [["init",2,1000],["is_allowed","clientA",100],["is_allowed","clientA",200],["is_allowed","clientA",300],["is_allowed","clientA",1101],["is_allowed","clientB",300]]

Output:

[null,true,true,false,true,true]

Worked solution and explanation

What this really is

Strip the API-gateway costume and this is a per-client sliding window with two decisions that separate candidates. Anyone can keep a deque of accepted timestamps per client; the interview lives in two places. First, the eviction boundary: the window is half-open, so a call exactly window_ms old has already expired and must free a slot. Write the eviction with > instead of >= and a client calling exactly one window after its oldest accepted call is wrongly denied. Second, denials leave no trace: record clientA's rejected call at t=300 and its call at t=1101 comes back False, and a client that keeps retrying keeps pushing its own cooldown out forever. All of it has to live per client, or clientA's burst throttles clientB.

Break down the requirements

Step 1: Match the operations contract exactly

You drive a list of operations. The first entry ['init', max_requests, window_ms] configures the limiter (milliseconds, integers) and contributes None. Every later ['is_allowed', client_id, timestamp] op contributes True or False. The function returns one result per op, in order, so the results list lines up index-for-index with operations.

Step 2: Isolate state per client

A defaultdict(deque) keyed by client_id gives each client an independent timestamp queue without an explicit existence check. Sharing a single deque across clients lets one chatty client throttle everyone else, which the visible test catches: clientB at t=300 is allowed because its own deque is empty even though clientA is currently throttled.

Step 3: Evict in milliseconds, deny without recording

Pop from the head while timestamp - q[0] >= window_ms. The window is half-open: a stored call exactly window_ms behind the new timestamp falls outside and frees a slot. Then if len(q) < max_requests, append and return True; otherwise return False without recording anything.

The solution

Per-client sliding window in milliseconds
from collections import defaultdict, deque


class RateLimiter:
    def __init__(self, max_requests, window_ms):
        self.max_requests = max_requests
        self.window_ms = window_ms
        self.clients = defaultdict(deque)

    def is_allowed(self, client_id, timestamp):
        q = self.clients[client_id]
        while q and timestamp - q[0] >= self.window_ms:
            q.popleft()
        if len(q) < self.max_requests:
            q.append(timestamp)
            return True
        return False


def run_rate_limiter(operations):
    limiter = None
    results = []
    for op in operations:
        method, args = op[0], op[1:]
        if method == "init":
            limiter = RateLimiter(*args)
            results.append(None)
        else:
            results.append(limiter.is_allowed(*args))
    return results

Common follow-up questions