The Velvet Rope

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

Domain
Python
Difficulty
medium
Seniority
L4
Asked in
Revolut

Problem

A request gateway throttles each client independently: within any window of period_seconds, one client may have at most max_calls requests admitted. Build a RateLimiter(max_calls, period_seconds) whose allow(user_id, now) admits a request only when that client has fewer than max_calls admitted requests still inside the trailing window that ends at now and reaches back period_seconds; a request that landed exactly period_seconds ago has already aged out, and a rejected request is never recorded against the client. The harness replays the calls through RateLimiter_driver(label, max_calls, period_seconds, calls), which builds one limiter and feeds each [user_id, timestamp] pair from calls to allow in order, returning the list of decisions.

Some users get in. Others wait outside until the window resets.

Example 1

Input:

max_calls = 3, period_seconds = 10, calls = [["userA",0],["userA",3],["userA",6],["userA",9]], label = "allow_sequence"

Output:

[true,true,true,false]

Example 2

Input:

max_calls = 2, period_seconds = 10, calls = [["userA",0],["userA",5],["userA",9],["userA",11]], label = "window_reset"

Output:

[true,true,false,true]

Worked solution and explanation

What this really is under the costume

Strip the gateway story away and this is a per-key sliding-window log: for each user_id, a deque of its own admitted timestamps, trimmed from the front as they age out. A plain counter per user cannot do it, because a counter only grows and never learns when a slot frees up; one shared history cannot do it either, because one client's burst would throttle everyone. Everyone reaches for a dict of user to timestamps. What separates candidates is two quiet details. First, the window is half-open, so an admit sitting exactly period_seconds in the past must drop out before you count. Second, only admitted calls are recorded, so a rejection must leave that user's state untouched. Miss either and the client is throttled harder than the contract allows, and every later decision for that user drifts with it.

Break down the requirements

Step 1: Track admitted timestamps per user

Each user has an independent window. Key a dict by user_id to a deque of that user's admitted timestamps so one client's traffic never bleeds into another's.

Step 2: Expire the front of the window first

Before each decision, pop from the front every recorded timestamp at or before now - period_seconds. The window is half-open, so a timestamp equal to now - period_seconds is already expired and goes too. A deque makes this front-popping O(1).

Step 3: Decide on the surviving count

If fewer than max_calls timestamps remain after expiry, admit the call and append now; otherwise reject and append nothing. The append-only-on-admit rule is the whole reason rejected bursts cannot starve the user forever: on window_reset, recording the rejected call at 9.0 would keep the window full and wrongly turn away the call at 11.0.

Step 4: Wire the driver entrypoint

The grader calls one top-level function, so RateLimiter_driver constructs a single limiter and replays every [user_id, timestamp] pair through allow, collecting the booleans. The label argument is just a test tag; it does not affect the result.

The solution

Per-user deque with half-open window expiry
from collections import deque


class RateLimiter:
    def __init__(self, max_calls: int, period_seconds: float):
        self.max_calls = max_calls
        self.period = period_seconds
        self.users = {}

    def allow(self, user_id: str, now: float) -> bool:
        if user_id not in self.users:
            self.users[user_id] = deque()
        dq = self.users[user_id]
        cutoff = now - self.period
        while dq and dq[0] <= cutoff:
            dq.popleft()
        if len(dq) < self.max_calls:
            dq.append(now)
            return True
        return False


def RateLimiter_driver(label, max_calls, period_seconds, calls):
    limiter = RateLimiter(max_calls, period_seconds)
    return [limiter.allow(user_id, timestamp) for user_id, timestamp in calls]

Common follow-up questions