api_base

Base classes and utilities for external API services.

This module provides common functionality for all API providers including: - Rate limiting with moving window approach - Common type definitions - Base scoring and normalization methods - Shared utilities for API interactions

ScoredRelease

Bases: TypedDict

Type definition for a scored release with metadata and scoring details.

ApiRateLimiter

ApiRateLimiter(requests_per_window, window_seconds)

Advanced rate limiter using a moving window approach for API calls.

This rate limiter tracks API calls within a sliding time window to ensure compliance with API rate limits. It uses a moving window algorithm that's more accurate than simple token bucket approaches.

Attributes:

Name	Type	Description
requests_per_window		Maximum number of requests allowed in the time window
window_seconds		Size of the time window in seconds
call_times	list[float]	List of timestamps for recent API calls
lock		Asyncio lock for thread-safe operations

Initialize the rate limiter.

Parameters:

Name	Type	Description	Default
requests_per_window	int	Maximum requests allowed in the time window	required
window_seconds	float	Duration of the time window in seconds	required

Raises:

Type	Description
ValueError	If parameters are not positive numbers

Source code in src/services/api/api_base.py

def __init__(self, requests_per_window: int, window_seconds: float) -> None:
    """Initialize the rate limiter.

    Args:
        requests_per_window: Maximum requests allowed in the time window
        window_seconds: Duration of the time window in seconds

    Raises:
        ValueError: If parameters are not positive numbers

    """
    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)

    self.requests_per_window = requests_per_window
    self.window_seconds = window_seconds
    self.call_times: list[float] = []
    self.lock = asyncio.Lock()
    self.total_requests = 0  # Track total requests made
    self.total_wait_time = 0.0  # Track cumulative wait time

acquire async

acquire()

Acquire permission to make an API call, waiting if necessary.

Returns:

Type	Description
float	The amount of time (in seconds) that was spent waiting

Source code in src/services/api/api_base.py

async def acquire(self) -> float:
    """Acquire permission to make an API call, waiting if necessary.

    Returns:
        The amount of time (in seconds) that was spent waiting

    """
    async with self.lock:
        wait_time = await self._wait_if_needed()
        self.call_times.append(time.monotonic())
        self.total_requests += 1
        self.total_wait_time += wait_time
        return wait_time

release

release()

Release method for compatibility (no-op for this implementation).

Source code in src/services/api/api_base.py

def release(self) -> None:
    """Release method for compatibility (no-op for this implementation)."""

get_stats

get_stats()

Get current rate limiter statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary containing current stats and configuration

Source code in src/services/api/api_base.py

def get_stats(self) -> dict[str, Any]:
    """Get current rate limiter statistics.

    Returns:
        Dictionary containing current stats and configuration

    """
    now = time.monotonic()
    cutoff = now - self.window_seconds
    current_calls = [t for t in self.call_times if t > cutoff]

    return {
        "requests_per_window": self.requests_per_window,
        "window_seconds": self.window_seconds,
        "current_calls_in_window": len(current_calls),
        "available_capacity": max(0, self.requests_per_window - len(current_calls)),
        "window_utilization": (len(current_calls) / self.requests_per_window if self.requests_per_window > 0 else 0),
        "total_requests": self.total_requests,
        "total_wait_time": self.total_wait_time,
        "avg_wait_time": self.total_wait_time / max(1, self.total_requests),
    }

BaseApiClient

BaseApiClient(console_logger, error_logger)

Base class for API client implementations.

Provides common functionality for all API clients including - Name normalization - Year validation - Common scoring logic

Initialize base API client.

Parameters:

Name	Type	Description	Default
console_logger	Logger	Logger for console output	required
error_logger	Logger	Logger for error messages	required

Source code in src/services/api/api_base.py

def __init__(self, console_logger: logging.Logger, error_logger: logging.Logger) -> None:
    """Initialize base API client.

    Args:
        console_logger: Logger for console output
        error_logger: Logger for error messages

    """
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.compilation_pattern = re.compile(
        r"\b(compilation|greatest\s+hits|best\s+of|collection|anthology)\b",
        re.IGNORECASE,
    )