How to Mock and Test a Rate-Limited Source Stream in Python

Build a class that rate-limits emitted items using a sliding window and test it with a simulated stream in Python.

Medium Python 3.9+ Aug 9, 2026 Big data & Spark 16 views 0 copies

rate-limiting mock-testing streaming sliding-window deque

Python code

28 lines
Python 3.9+
Run it 
import time
from collections import deque


class RateLimitedSource:
    def __init__(self, max_rate, window=1.0):
        self.max_rate = max_rate
        self.window = window
        self._timestamps = deque()

    def emit(self, item):
        now = time.monotonic()
        while self._timestamps and self._timestamps[0] <= now - self.window:
            self._timestamps.popleft()
        if len(self._timestamps) < self.max_rate:
            self._timestamps.append(now)
            return item
        raise RuntimeError("Rate limit exceeded")


if __name__ == "__main__":
    source = RateLimitedSource(max_rate=3, window=1.0)
    for i in range(5):
        try:
            print(source.emit(f"item-{i}"))
        except RuntimeError:
            print(f"item-{i} rejected")
            time.sleep(0.5)

Output

stdout 
item-0
item-1
item-2
item-3 rejected
item-4 rejected

How it works

The RateLimitedSource class uses time.monotonic() and a deque to track timestamps within a sliding window. Each emit call first removes timestamps older than the window, then checks if the count is below the max rate before allowing the item. When the limit is exceeded, a RuntimeError is raised, which the test loop catches to simulate a rejected event. This pattern mimics how streaming pipelines throttle data intake to prevent overload. The mock stream in __main__ demonstrates graceful handling of rejections by sleeping briefly before retrying.

Common mistakes

  • Using `time.time()` instead of `time.monotonic()` allows system clock changes to skew rate calculations
  • Forgetting to evict old timestamps before checking the count leads to inaccurate throttling
  • Letting the deque grow unbounded in production without periodic cleanup causes memory bloat

Variations

  1. Use a token bucket algorithm instead of a sliding window for burst handling
  2. Replace RuntimeError with a custom RateLimitExceeded exception for cleaner error handling

Real-world use cases

  • Mocking upstream API rate limits when building a Kafka producer that fetches from a restricted REST service.
  • Simulating throttled data sources in unit tests for Spark streaming ingestion jobs.
  • Validating backpressure logic in a data pipeline that consumes from a rate-limited partner feed.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Big data & Spark

Keep learning

Related tutorials and quizzes for this topic.

Tutorials

Quizzes