API Rate Limiter for Data Fetcher

Overview

Build a data fetching service that retrieves information from a simulated external API. The service must enforce rate limits to prevent overwhelming the API endpoint. Implement both synchronous and asynchronous data fetchers with appropriate rate limiting.

Requirements

Core Functionality

1. Synchronous Data Fetcher: Create a function fetch_user_data(user_id: int) -> dict that simulates fetching user data. The function should:

   • Accept a user ID as input
   • Return a dictionary with user information (you can use mock data)
   • Be rate-limited to a maximum of 5 calls per 10 seconds

2. Asynchronous Data Fetcher: Create an async function fetch_orders_async(order_id: int) -> dict that simulates fetching order data. The function should:

   • Accept an order ID as input
   • Return a dictionary with order information (you can use mock data)
   • Be rate-limited to a maximum of 10 calls per 15 seconds

3. Weighted API Call: Create a function fetch_bulk_report(report_size: str) -> dict that simulates fetching reports of different sizes. The function should:

   • Accept a report size parameter: "small", "medium", or "large"
   • Return mock report data
   • Be rate-limited with different weights:
     • "small" reports: weight of 1
     • "medium" reports: weight of 3
     • "large" reports: weight of 5
   • Total limit: 20 weight units per 20 seconds

Implementation Details

• Use appropriate rate limiting decorators to enforce the limits
• The rate limiter should block when limits are exceeded (default blocking behavior)
• Use an in-memory storage backend for simplicity
• Mock data can be simple dictionaries (e.g., {"user_id": user_id, "name": "User Name"})

Test Cases

Test Case 1: Synchronous Rate Limit { .test_case @test }

File: test_rate_limiter.py { .test_file }

Test that the synchronous fetcher enforces the 5 calls per 10 seconds limit:

import time
from solution import fetch_user_data

# Should succeed for first 5 calls
for i in range(5):
    result = fetch_user_data(i)
    assert result is not None
    assert "user_id" in result

# 6th call should block briefly, then succeed
start_time = time.time()
result = fetch_user_data(5)
elapsed = time.time() - start_time
assert result is not None
assert elapsed > 1  # Should have waited

Test Case 2: Async Rate Limit { .test_case @test }

File: test_rate_limiter.py { .test_file }

Test that the async fetcher enforces the 10 calls per 15 seconds limit:

import asyncio
from solution import fetch_orders_async

async def test_async_limit():
    # Should succeed for first 10 calls
    for i in range(10):
        result = await fetch_orders_async(i)
        assert result is not None
        assert "order_id" in result

    # 11th call should block briefly, then succeed
    start_time = asyncio.get_event_loop().time()
    result = await fetch_orders_async(10)
    elapsed = asyncio.get_event_loop().time() - start_time
    assert result is not None
    assert elapsed > 1  # Should have waited

asyncio.run(test_async_limit())

Test Case 3: Weighted Rate Limit { .test_case @test }

File: test_rate_limiter.py { .test_file }

Test that weighted reports consume appropriate capacity:

from solution import fetch_bulk_report

# Fetch 4 small reports (4 weight units total)
for i in range(4):
    result = fetch_bulk_report("small")
    assert result is not None

# Fetch 3 medium reports (9 weight units total)
for i in range(3):
    result = fetch_bulk_report("medium")
    assert result is not None

# Total: 13 weight units used
# Fetch 1 large report (5 weight units) - should succeed (18 total)
result = fetch_bulk_report("large")
assert result is not None

# Try another medium report (3 weight units) - would exceed 20, should block
import time
start_time = time.time()
result = fetch_bulk_report("medium")
elapsed = time.time() - start_time
assert result is not None
assert elapsed > 1  # Should have waited

Dependencies { .dependencies }

pyrate-limiter { .dependency }

Provides rate limiting capabilities using the leaky bucket algorithm.

Deliverables

• solution.py: Contains the three rate-limited functions
• test_rate_limiter.py: Contains test cases validating the rate limiting behavior