This commit is contained in:
Harivansh Rathi 2025-12-14 17:15:37 -05:00
parent 76062021e9
commit c548626d7f
13 changed files with 570 additions and 1 deletions

View file

@ -0,0 +1,49 @@
"""
Rate Limiter
You're building an API gateway for a SaaS platform. To prevent abuse
and ensure fair usage, you need to implement a rate limiter that tracks
requests per user using a sliding window algorithm.
The limiter should allow at most `max_requests` per user within any
`window_seconds` time period.
Example 1:
limiter = RateLimiter(max_requests=3, window_seconds=60)
limiter.allow_request("user1", timestamp=0) # True (1st request)
limiter.allow_request("user1", timestamp=30) # True (2nd request)
limiter.allow_request("user1", timestamp=45) # True (3rd request)
limiter.allow_request("user1", timestamp=50) # False (limit reached)
limiter.allow_request("user1", timestamp=61) # True (1st expired)
Example 2:
limiter = RateLimiter(max_requests=2, window_seconds=10)
limiter.allow_request("user1", timestamp=0) # True
limiter.allow_request("user2", timestamp=0) # True (different user)
limiter.allow_request("user1", timestamp=5) # True
limiter.allow_request("user1", timestamp=8) # False
Constraints:
- max_requests >= 1
- window_seconds >= 1
- Timestamps are non-negative integers (seconds)
- Timestamps are non-decreasing per user
- user_id is a non-empty string
"""
class RateLimiter:
"""Sliding window rate limiter for API request throttling."""
def __init__(self, max_requests: int, window_seconds: int):
"""Initialize with request limit and time window."""
pass # Your implementation here
def allow_request(self, user_id: str, timestamp: int) -> bool:
"""Return True if request allowed, False if rate limited."""
pass # Your implementation here
def get_remaining(self, user_id: str, timestamp: int) -> int:
"""Return remaining requests allowed for user at timestamp."""
pass # Your implementation here

View file

@ -0,0 +1,70 @@
"""Tests for rate-limiter."""
import pytest
from solution import RateLimiter
class TestBasicCases:
"""Test basic functionality with typical inputs."""
def test_allow_within_limit(self):
"""Test requests within the limit are allowed."""
limiter = RateLimiter(max_requests=3, window_seconds=60)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 30) == True
assert limiter.allow_request("user1", 45) == True
def test_block_over_limit(self):
"""Test requests over limit are blocked."""
limiter = RateLimiter(max_requests=2, window_seconds=60)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 30) == True
assert limiter.allow_request("user1", 45) == False
def test_multiple_users_independent(self):
"""Test each user has independent limits."""
limiter = RateLimiter(max_requests=1, window_seconds=60)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user2", 0) == True
assert limiter.allow_request("user1", 30) == False
assert limiter.allow_request("user2", 30) == False
def test_get_remaining_basic(self):
"""Test remaining count decreases with requests."""
limiter = RateLimiter(max_requests=3, window_seconds=60)
assert limiter.get_remaining("user1", 0) == 3
limiter.allow_request("user1", 0)
assert limiter.get_remaining("user1", 0) == 2
class TestEdgeCases:
"""Test edge cases and boundary conditions."""
def test_window_expiration(self):
"""Test old requests expire from window."""
limiter = RateLimiter(max_requests=2, window_seconds=60)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 30) == True
assert limiter.allow_request("user1", 45) == False
assert limiter.allow_request("user1", 61) == True
def test_single_request_limit(self):
"""Test with limit of 1 request."""
limiter = RateLimiter(max_requests=1, window_seconds=10)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 5) == False
assert limiter.allow_request("user1", 11) == True
def test_new_user_full_allowance(self):
"""Test new users start with full allowance."""
limiter = RateLimiter(max_requests=5, window_seconds=60)
limiter.allow_request("user1", 0)
assert limiter.get_remaining("new_user", 20) == 5
def test_rapid_same_timestamp(self):
"""Test multiple requests at same timestamp."""
limiter = RateLimiter(max_requests=3, window_seconds=1)
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 0) == True
assert limiter.allow_request("user1", 0) == False