API Request Tracker

A simple API request tracking system that enforces rate limits on incoming requests.

Overview

Build a request tracker that manages API request limits for different services. The tracker should enforce rate limits and provide feedback about whether requests can be processed or should be delayed.

Capabilities

Enforces basic rate limits

• When processing requests under the rate limit, all requests are accepted @test
• When exceeding the rate limit in non-blocking mode, requests are rejected @test

Handles blocking behavior

• When operating in blocking mode, requests wait until the rate limit allows them to proceed @test

Respects timeout configuration

• When blocking with a timeout, requests fail if they cannot be processed within the timeout period @test

Requirements

The tracker should:

1. Accept a rate limit configuration (e.g., 5 requests per second)
2. Track requests by name/identifier
3. Support both blocking and non-blocking operation modes
4. Allow configurable timeouts for blocking requests
5. Return boolean values indicating whether a request can proceed

Implementation

@generates

API

class RequestTracker:
    """
    Tracks API requests and enforces rate limits.

    Args:
        max_requests: Maximum number of requests allowed
        time_window_seconds: Time window for the rate limit in seconds
    """
    def __init__(self, max_requests: int, time_window_seconds: int):
        pass

    def process_request(
        self,
        request_name: str,
        blocking: bool = True,
        timeout: float | None = None
    ) -> bool:
        """
        Process a request and enforce rate limits.

        Args:
            request_name: Identifier for the request
            blocking: If True, wait for rate limit to allow request.
                     If False, return immediately if rate limit exceeded.
            timeout: Maximum time to wait in blocking mode (seconds).
                    None means wait indefinitely.

        Returns:
            True if request can proceed, False if rejected due to rate limit
        """
        pass

    def close(self):
        """Clean up resources."""
        pass

Dependencies { .dependencies }

pyrate-limiter { .dependency }

Provides rate limiting functionality.