batch_fetcher

Batch track fetching module.

This module handles fetching tracks from Music.app in batches to avoid timeouts when processing large libraries.

BatchTrackFetcher

BatchTrackFetcher(
    ap_client,
    cache_service,
    console_logger,
    error_logger,
    config,
    *,
    track_validator,
    artist_processor,
    snapshot_loader,
    snapshot_persister,
    can_use_snapshot,
    dry_run=False,
    analytics=None
)

Fetches tracks from Music.app in batches.

This class handles: - Batch-based track fetching to avoid AppleScript timeouts - Parse failure tracking and recovery - Caching and snapshot persistence of fetched tracks

Initialize the batch track fetcher.

Parameters:

Name	Type	Description	Default
ap_client	AppleScriptClientProtocol	AppleScript client for Music.app communication	required
cache_service	CacheServiceProtocol	Cache service for storing fetched tracks	required
console_logger	Logger	Logger for info/debug messages	required
error_logger	Logger	Logger for error messages	required
config	AppConfig	Typed application configuration	required
track_validator	Callable[[list[TrackDict]], list[TrackDict]]	Callback to validate tracks for security	required
artist_processor	Callable[[list[TrackDict]], Awaitable[None]]	Async callback to process artist renames	required
snapshot_loader	Callable[[], Awaitable[list[TrackDict] | None]]	Async callback to load tracks from snapshot	required
snapshot_persister	Callable[[list[TrackDict], list[str] | None], Awaitable[None]]	Async callback to persist tracks to snapshot	required
can_use_snapshot	Callable[[str | None], bool]	Callback to check if snapshot can be used	required
dry_run	bool	Whether running in dry-run mode	False
analytics	AnalyticsProtocol | None	Optional analytics instance for batch mode logging	None

Source code in src/core/tracks/batch_fetcher.py

def __init__(
    self,
    ap_client: AppleScriptClientProtocol,
    cache_service: CacheServiceProtocol,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    config: AppConfig,
    *,
    track_validator: Callable[[list[TrackDict]], list[TrackDict]],
    artist_processor: Callable[[list[TrackDict]], Awaitable[None]],
    snapshot_loader: Callable[[], Awaitable[list[TrackDict] | None]],
    snapshot_persister: Callable[[list[TrackDict], list[str] | None], Awaitable[None]],
    can_use_snapshot: Callable[[str | None], bool],
    dry_run: bool = False,
    analytics: AnalyticsProtocol | None = None,
) -> None:
    """Initialize the batch track fetcher.

    Args:
        ap_client: AppleScript client for Music.app communication
        cache_service: Cache service for storing fetched tracks
        console_logger: Logger for info/debug messages
        error_logger: Logger for error messages
        config: Typed application configuration
        track_validator: Callback to validate tracks for security
        artist_processor: Async callback to process artist renames
        snapshot_loader: Async callback to load tracks from snapshot
        snapshot_persister: Async callback to persist tracks to snapshot
        can_use_snapshot: Callback to check if snapshot can be used
        dry_run: Whether running in dry-run mode
        analytics: Optional analytics instance for batch mode logging
    """
    self.ap_client = ap_client
    self.cache_service = cache_service
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.config = config
    self._track_validator = track_validator
    self._artist_processor = artist_processor
    self._snapshot_loader = snapshot_loader
    self._snapshot_persister = snapshot_persister
    self._can_use_snapshot = can_use_snapshot
    self.dry_run = dry_run
    self.analytics = analytics

fetch_all_tracks async

fetch_all_tracks(
    batch_size=500, *, skip_snapshot_check=False
)

Fetch all tracks from Music.app in batches.

Parameters:

Name	Type	Description	Default
batch_size	int	Number of tracks to fetch per batch	500
skip_snapshot_check	bool	Skip snapshot validation (used when already validated upstream)	False

Returns:

Type	Description
list[TrackDict]	List of all track dictionaries

Source code in src/core/tracks/batch_fetcher.py

async def fetch_all_tracks(
    self,
    batch_size: int = 500,
    *,
    skip_snapshot_check: bool = False,
) -> list[TrackDict]:
    """Fetch all tracks from Music.app in batches.

    Args:
        batch_size: Number of tracks to fetch per batch
        skip_snapshot_check: Skip snapshot validation (used when already validated upstream)

    Returns:
        List of all track dictionaries
    """
    # Try loading from snapshot first
    if not skip_snapshot_check and self._can_use_snapshot(None):
        snapshot_tracks = await self._snapshot_loader()
        if snapshot_tracks is not None:
            self.console_logger.info(
                "\u2713 Loaded %d tracks from snapshot cache; skipping batch fetch",
                len(snapshot_tracks),
            )
            return snapshot_tracks

    # Snapshot not available - proceed with batch processing
    all_tracks = await self._fetch_tracks_in_batches(batch_size)

    # Cache and persist results
    await self._cache_and_persist_results(all_tracks)

    return all_tracks