rate_limiter

AppleScript rate limiter module.

This module provides rate limiting for AppleScript execution using a moving window approach with concurrency control.

RateLimiterStats

Bases: TypedDict

Statistics from the AppleScript rate limiter.

AppleScriptRateLimiter

AppleScriptRateLimiter(
    requests_per_window,
    window_seconds,
    max_concurrent=3,
    logger=None,
)

Advanced rate limiter using a moving window approach.

Initialize the rate limiter with configurable limits.

Parameters:

Name	Type	Description	Default
requests_per_window	int	Maximum requests allowed per time window.	required
window_seconds	float	Duration of the sliding window in seconds.	required
max_concurrent	int	Maximum concurrent requests (semaphore limit).	3
logger	Logger | None	Optional logger instance for debug output.	None

Raises:

Type	Description
ValueError	If any numeric parameter is not positive.

Source code in src/services/apple/rate_limiter.py

def __init__(
    self,
    requests_per_window: int,
    window_seconds: float,
    max_concurrent: int = 3,
    logger: logging.Logger | None = None,
) -> None:
    """Initialize the rate limiter with configurable limits.

    Args:
        requests_per_window: Maximum requests allowed per time window.
        window_seconds: Duration of the sliding window in seconds.
        max_concurrent: Maximum concurrent requests (semaphore limit).
        logger: Optional logger instance for debug output.

    Raises:
        ValueError: If any numeric parameter is not positive.

    """
    if requests_per_window <= 0:
        msg = "requests_per_window must be a positive integer"
        raise ValueError(msg)
    if window_seconds <= 0:
        msg = "window_seconds must be a positive number"
        raise ValueError(msg)
    if max_concurrent <= 0:
        msg = "max_concurrent must be a positive integer"
        raise ValueError(msg)

    self.requests_per_window = requests_per_window
    self.window_seconds = window_seconds
    self.request_timestamps: deque[float] = deque()
    self.semaphore: asyncio.Semaphore | None = None
    self.max_concurrent = max_concurrent
    self.logger = logger or logging.getLogger(__name__)
    self.total_requests: int = 0
    self.total_wait_time: float = 0.0

initialize async

initialize()

Initialize the rate limiter for async operation.

Creates the asyncio semaphore for concurrency control. Must be called before using acquire/release in an async context.

Source code in src/services/apple/rate_limiter.py

async def initialize(self) -> None:
    """Initialize the rate limiter for async operation.

    Creates the asyncio semaphore for concurrency control. Must be called
    before using acquire/release in an async context.

    """
    if self.semaphore is None:
        try:
            self.semaphore = asyncio.Semaphore(self.max_concurrent)
            self.logger.debug("RateLimiter initialized with max_concurrent: %s", self.max_concurrent)
            # Yield control to event loop to make this properly async
            await asyncio.sleep(0)
        except (ValueError, TypeError, RuntimeError, asyncio.InvalidStateError) as e:
            self.logger.exception("Error initializing RateLimiter semaphore: %s", e)
            raise

acquire async

acquire()

Acquire permission to make a request.

Waits if necessary to respect rate limits, then acquires the semaphore. Records the request timestamp for sliding window calculation.

Returns:

Type	Description
float	Wait time in seconds (0.0 if no wait was needed).

Source code in src/services/apple/rate_limiter.py

async def acquire(self) -> float:
    """Acquire permission to make a request.

    Waits if necessary to respect rate limits, then acquires the semaphore.
    Records the request timestamp for sliding window calculation.

    Returns:
        Wait time in seconds (0.0 if no wait was needed).

    """
    if self.semaphore is None:
        msg = "RateLimiter not initialized"
        raise RuntimeError(msg)
    rate_limit_wait_time = await self._wait_if_needed()
    self.total_requests += 1
    self.total_wait_time += rate_limit_wait_time
    await self.semaphore.acquire()
    return rate_limit_wait_time

release

release()

Release the semaphore after request completion.

Should be called after each request finishes to allow other concurrent requests to proceed.

Source code in src/services/apple/rate_limiter.py

def release(self) -> None:
    """Release the semaphore after request completion.

    Should be called after each request finishes to allow other
    concurrent requests to proceed.

    """
    if self.semaphore is None:
        return
    self.semaphore.release()

get_stats

get_stats()

Get current rate limiter statistics.

Returns:

Type	Description
RateLimiterStats	Dictionary containing:
RateLimiterStats	"total_requests": Total requests processed
RateLimiterStats	"total_wait_time": Cumulative wait time in seconds
RateLimiterStats	"avg_wait_time": Average wait time per request
RateLimiterStats	"current_calls_in_window": Current requests within sliding window
RateLimiterStats	"requests_per_window": Configured request limit
RateLimiterStats	"window_seconds": Window duration in seconds

Source code in src/services/apple/rate_limiter.py

def get_stats(self) -> RateLimiterStats:
    """Get current rate limiter statistics.

    Returns:
        Dictionary containing:
        - "total_requests": Total requests processed
        - "total_wait_time": Cumulative wait time in seconds
        - "avg_wait_time": Average wait time per request
        - "current_calls_in_window": Current requests within sliding window
        - "requests_per_window": Configured request limit
        - "window_seconds": Window duration in seconds

    """
    now = time.monotonic()
    self.request_timestamps = deque(ts for ts in self.request_timestamps if now - ts <= self.window_seconds)
    return {
        "total_requests": self.total_requests,
        "total_wait_time": self.total_wait_time,
        "avg_wait_time": self.total_wait_time / max(1, self.total_requests),
        "current_calls_in_window": len(self.request_timestamps),
        "requests_per_window": self.requests_per_window,
        "window_seconds": self.window_seconds,
    }