software/src/orchestration/rate_limiter.py

1"""Rate limiter for coordinating parallel API calls.23Provides thread-safe rate limiting to prevent overwhelming4the Claude API when multiple book workers are running in parallel.5"""67from __future__ import annotations89import threading10import time11from contextlib import contextmanager12from typing import Generator131415class SharedRateLimiter:16    """Thread-safe rate limiter using a bounded semaphore.1718    Coordinates API access across multiple worker threads to prevent19    rate limiting errors. Uses a BoundedSemaphore to limit concurrent20    requests and adds a delay between releases.2122    Example:23        limiter = SharedRateLimiter(max_concurrent=5, delay=1.2)2425        # In worker thread:26        with limiter:27            call_claude_api()2829    Attributes:30        max_concurrent: Maximum concurrent API calls allowed.31        delay: Delay in seconds between API calls.32    """3334    def __init__(35        self,36        max_concurrent: int = 5,37        delay: float = 1.2,38    ) -> None:39        """Initialize the rate limiter.4041        Args:42            max_concurrent: Maximum concurrent API calls (default: 5).43            delay: Minimum delay between calls in seconds (default: 1.2).44        """45        self.max_concurrent = max_concurrent46        self.delay = delay4748        # Create semaphore for limiting concurrent calls49        self._semaphore = threading.BoundedSemaphore(max_concurrent)5051        # Create lock for coordinating delays52        self._delay_lock = threading.Lock()5354        # Track last call time55        self._last_call_time = 0.05657    def acquire(self) -> None:58        """Acquire a rate limit slot.5960        Blocks until a slot is available and the minimum delay has passed61        since the last call.62        """63        # First, acquire semaphore slot64        self._semaphore.acquire()6566        # Then, ensure minimum delay between calls67        with self._delay_lock:68            now = time.time()69            elapsed = now - self._last_call_time70            if elapsed < self.delay:71                time.sleep(self.delay - elapsed)72            self._last_call_time = time.time()7374    def release(self) -> None:75        """Release a rate limit slot."""76        self._semaphore.release()7778    @contextmanager79    def __call__(self) -> Generator[None, None, None]:80        """Context manager for rate limiting.8182        Yields:83            None, but blocks until slot is available.84        """85        self.acquire()86        try:87            yield88        finally:89            self.release()9091    def __enter__(self) -> SharedRateLimiter:92        """Enter context manager."""93        self.acquire()94        return self9596    def __exit__(self, *args: object) -> None:97        """Exit context manager."""98        self.release()99100    def shutdown(self) -> None:101        """Clean up resources (no-op for thread-based limiter)."""102        pass103104105class DummyRateLimiter:106    """No-op rate limiter for single-process usage or testing.107108    Provides the same interface as SharedRateLimiter but does nothing.109    Useful for testing or when rate limiting is not needed.110    """111112    def __init__(113        self,114        max_concurrent: int = 5,115        delay: float = 1.2,116    ) -> None:117        """Initialize the dummy rate limiter.118119        Args:120            max_concurrent: Ignored (for API compatibility).121            delay: Delay in seconds between calls (still applied).122        """123        self.max_concurrent = max_concurrent124        self.delay = delay125        self._last_call_time = 0.0126127    def acquire(self) -> None:128        """Acquire rate limit slot (applies delay only)."""129        now = time.time()130        elapsed = now - self._last_call_time131        if elapsed < self.delay:132            time.sleep(self.delay - elapsed)133        self._last_call_time = time.time()134135    def release(self) -> None:136        """Release rate limit slot (no-op)."""137        pass138139    @contextmanager140    def __call__(self) -> Generator[None, None, None]:141        """Context manager for rate limiting."""142        self.acquire()143        try:144            yield145        finally:146            self.release()147148    def __enter__(self) -> DummyRateLimiter:149        """Enter context manager."""150        self.acquire()151        return self152153    def __exit__(self, *args: object) -> None:154        """Exit context manager."""155        self.release()156157    def shutdown(self) -> None:158        """Clean up resources (no-op)."""159        pass160