types

Type definitions for the utils layer.

This module provides type imports that utils modules can use without depending on the services layer, maintaining clean architecture boundaries.

CachedApiResult

Bases: BaseModel

Cached API result.

model_dump

model_dump(**_kwargs)

Export model as dictionary (Pydantic v2 compatibility).

Source code in src/core/models/track_models.py

def model_dump(self, **_kwargs: Any) -> dict[str, Any]:
    """Export model as dictionary (Pydantic v2 compatibility)."""
    return {
        "artist": self.artist,
        "album": self.album,
        "year": self.year,
        "source": self.source,
        "timestamp": self.timestamp,
        "ttl": self.ttl,
        "metadata": self.metadata,
        "api_response": self.api_response,
    }

TrackDict

Bases: BaseModel

Track information from Apple Music.

get

get(key, default=None)

Get attribute value with default, mimicking dict.get() behavior.

Parameters:

Name	Type	Description	Default
key	str	The attribute name to retrieve	required
default	TrackFieldValue	The default value to return if attribute doesn't exist	None

Returns:

Type	Description
TrackFieldValue	The attribute value if it exists, otherwise the default value

Source code in src/core/models/track_models.py

def get(self, key: str, default: TrackFieldValue = None) -> TrackFieldValue:
    """Get attribute value with default, mimicking dict.get() behavior.

    Args:
        key: The attribute name to retrieve
        default: The default value to return if attribute doesn't exist

    Returns:
        The attribute value if it exists, otherwise the default value

    """
    try:
        # First, try to get the field value using getattr
        value = getattr(self, key)
    except AttributeError:
        # For Pydantic v1 with extra="allow", additional fields are stored in __dict__
        if key in self.__dict__:
            value = self.__dict__[key]
            return default if value is None and default is not None else value
        return default

    # Return the value, but if it's None, and we have a non-None default, use default
    return default if value is None and default is not None else value

copy

copy(**kwargs)

Create a copy of the TrackDict with optional field updates.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Fields to update in the copy	{}

Returns:

Type	Description
TrackDict	A new TrackDict instance with updated fields

Source code in src/core/models/track_models.py

def copy(self, **kwargs: Any) -> TrackDict:
    """Create a copy of the TrackDict with optional field updates.

    Args:
        **kwargs: Fields to update in the copy

    Returns:
        A new TrackDict instance with updated fields

    """
    # Pydantic v2 model_dump() includes extra fields automatically
    data = self.model_dump()
    data.update(kwargs)
    return TrackDict(**data)

AnalyticsProtocol

Bases: Protocol

Protocol for analytics services providing performance tracking.

Defines the contract for executing wrapped function calls (sync and async) with analytics instrumentation, and batch mode for suppressing console output during heavy processing.

execute_async_wrapped_call async

execute_async_wrapped_call(
    func, event_type, *args, **kwargs
)

Execute an async function with analytics tracking.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Async function to execute	required
event_type	str	Type of event for tracking	required
*args	Any	Function arguments	()
**kwargs	Any	Function keyword arguments	{}

Source code in src/core/models/protocols.py

async def execute_async_wrapped_call(
    self,
    func: Callable[..., Any],
    event_type: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Execute an async function with analytics tracking.

    Args:
        func: Async function to execute
        event_type: Type of event for tracking
        *args: Function arguments
        **kwargs: Function keyword arguments

    """
    ...

execute_sync_wrapped_call

execute_sync_wrapped_call(
    func, event_type, *args, **kwargs
)

Execute a sync function with analytics tracking.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	Sync function to execute	required
event_type	str	Type of event for tracking	required
*args	Any	Function arguments	()
**kwargs	Any	Function keyword arguments	{}

Source code in src/core/models/protocols.py

def execute_sync_wrapped_call(
    self,
    func: Callable[..., Any],
    event_type: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Execute a sync function with analytics tracking.

    Args:
        func: Sync function to execute
        event_type: Type of event for tracking
        *args: Function arguments
        **kwargs: Function keyword arguments

    """
    ...

batch_mode

batch_mode(message='Processing...', console=None)

Context manager that suppresses console logging and shows a spinner.

Parameters:

Name	Type	Description	Default
message	str	Message to display next to the spinner	'Processing...'
console	Console | None	Optional Rich Console instance	None

Returns:

Type	Description
AbstractAsyncContextManager[Status]	Async context manager yielding a Rich Status object

Source code in src/core/models/protocols.py

def batch_mode(
    self,
    message: str = "Processing...",
    console: Console | None = None,
) -> AbstractAsyncContextManager[Status]:
    """Context manager that suppresses console logging and shows a spinner.

    Args:
        message: Message to display next to the spinner
        console: Optional Rich Console instance

    Returns:
        Async context manager yielding a Rich Status object

    """
    ...

AppleScriptClientProtocol

Bases: Protocol

Protocol defining the interface for AppleScript clients.

This protocol allows both real and dry-run implementations to be used interchangeably for executing AppleScript commands.

initialize async

initialize()

Initialize the AppleScript client.

Source code in src/core/models/protocols.py

async def initialize(self) -> None:
    """Initialize the AppleScript client."""
    ...

run_script async

run_script(
    script_name,
    arguments=None,
    timeout=None,
    context_artist=None,
    context_album=None,
    context_track=None,
    label=None,
)

Run an AppleScript file by name.

Parameters:

Name	Type	Description	Default
script_name	str	Name of the script file to execute	required
arguments	list[str] | None	Optional arguments to pass to the script	None
timeout	float | None	Optional timeout in seconds	None
context_artist	str | None	Artist name for contextual logging (optional)	None
context_album	str | None	Album name for contextual logging (optional)	None
context_track	str | None	Track name for contextual logging (optional)	None
label	str | None	Custom label for logging (defaults to script_name)	None

Returns:

Type	Description
str | None	Script output or None if no output

Source code in src/core/models/protocols.py

async def run_script(
    self,
    script_name: str,
    arguments: list[str] | None = None,
    timeout: float | None = None,
    context_artist: str | None = None,
    context_album: str | None = None,
    context_track: str | None = None,
    label: str | None = None,
) -> str | None:
    """Run an AppleScript file by name.

    Args:
        script_name: Name of the script file to execute
        arguments: Optional arguments to pass to the script
        timeout: Optional timeout in seconds
        context_artist: Artist name for contextual logging (optional)
        context_album: Album name for contextual logging (optional)
        context_track: Track name for contextual logging (optional)
        label: Custom label for logging (defaults to script_name)

    Returns:
        Script output or None if no output

    """
    ...

fetch_tracks_by_ids async

fetch_tracks_by_ids(
    track_ids, batch_size=1000, timeout=None
)

Fetch tracks by their IDs using fetch_tracks_by_ids.applescript.

Parameters:

Name	Type	Description	Default
track_ids	list[str]	List of track IDs to fetch	required
batch_size	int	Maximum number of IDs per batch (default: 1000)	1000
timeout	float | None	Timeout in seconds for script execution	None

Returns:

Type	Description
list[dict[str, str]]	List of track dictionaries with metadata

Source code in src/core/models/protocols.py

async def fetch_tracks_by_ids(
    self,
    track_ids: list[str],
    batch_size: int = 1000,
    timeout: float | None = None,
) -> list[dict[str, str]]:
    """Fetch tracks by their IDs using fetch_tracks_by_ids.applescript.

    Args:
        track_ids: List of track IDs to fetch
        batch_size: Maximum number of IDs per batch (default: 1000)
        timeout: Timeout in seconds for script execution

    Returns:
        List of track dictionaries with metadata

    """
    ...

fetch_all_track_ids async

fetch_all_track_ids(timeout=None)

Fetch just track IDs from Music.app (lightweight operation).

This is used by Smart Delta to detect new/removed tracks without fetching full metadata.

Parameters:

Name	Type	Description	Default
timeout	float | None	Timeout in seconds for script execution	None

Returns:

Type	Description
list[str]	List of track ID strings

Source code in src/core/models/protocols.py

async def fetch_all_track_ids(self, timeout: float | None = None) -> list[str]:
    """Fetch just track IDs from Music.app (lightweight operation).

    This is used by Smart Delta to detect new/removed tracks without
    fetching full metadata.

    Args:
        timeout: Timeout in seconds for script execution

    Returns:
        List of track ID strings

    """
    ...

CacheServiceProtocol

Bases: Protocol

Protocol defining the interface for cache services.

This protocol defines methods for both in-memory caching with TTL support and persistent CSV-based caching for album information.

initialize async

initialize()

Initialize the cache service, loading any persistent data.

Source code in src/core/models/protocols.py

async def initialize(self) -> None:
    """Initialize the cache service, loading any persistent data."""
    ...

set

set(key_data, value, ttl=None)

Set a value in the cache with optional TTL.

Parameters:

Name	Type	Description	Default
key_data	CacheableKey	Key for the cached value (will be hashed if needed)	required
value	CacheableValue	Value to cache	required
ttl	int | None	Time-to-live in seconds (optional)	None

Source code in src/core/models/protocols.py

def set(self, key_data: CacheableKey, value: CacheableValue, ttl: int | None = None) -> None:
    """Set a value in the cache with optional TTL.

    Args:
        key_data: Key for the cached value (will be hashed if needed)
        value: Value to cache
        ttl: Time-to-live in seconds (optional)

    """
    ...

set_async async

set_async(key_data, value, ttl=None)

Asynchronously set a value in the cache with optional TTL.

Parameters:

Name	Type	Description	Default
key_data	CacheableKey	Key for the cached value (will be hashed if needed)	required
value	CacheableValue	Value to cache	required
ttl	int | None	Time-to-live in seconds (optional)	None

Source code in src/core/models/protocols.py

async def set_async(
    self,
    key_data: CacheableKey,
    value: CacheableValue,
    ttl: int | None = None,
) -> None:
    """Asynchronously set a value in the cache with optional TTL.

    Args:
        key_data: Key for the cached value (will be hashed if needed)
        value: Value to cache
        ttl: Time-to-live in seconds (optional)

    """
    ...

get_async async

get_async(
    key_data: Literal["ALL"], compute_func: None = None
) -> list[TrackDict]

get_async(
    key_data: str, compute_func: None = None
) -> list[TrackDict] | None

get_async(
    key_data: CacheableKey,
    compute_func: (
        Callable[[], Future[CacheableValue]] | None
    ) = None,
) -> CacheableValue

get_async(key_data, compute_func=None)

Get a value from cache, optionally computing it if not present.

Parameters:

Name	Type	Description	Default
key_data	CacheableKey	Key for the cached value ("ALL" for all tracks)	required
compute_func	Callable[[], Future[CacheableValue]] | None	Optional compute function to calculate value if not cached	None

Returns:

Type	Description
list[TrackDict] | CacheableValue	Cached value or computed value from factory, or None if not found

Source code in src/core/models/protocols.py

async def get_async(
    self,
    key_data: CacheableKey,
    compute_func: Callable[[], asyncio.Future[CacheableValue]] | None = None,
) -> list[TrackDict] | CacheableValue:
    """Get a value from cache, optionally computing it if not present.

    Args:
        key_data: Key for the cached value ("ALL" for all tracks)
        compute_func: Optional compute function to calculate value if not cached

    Returns:
        Cached value or computed value from factory, or None if not found

    """
    ...

invalidate

invalidate(key_data)

Invalidate (remove) a specific cache entry.

Parameters:

Name	Type	Description	Default
key_data	CacheableKey	Key of the cache entry to invalidate	required

Source code in src/core/models/protocols.py

def invalidate(self, key_data: CacheableKey) -> None:
    """Invalidate (remove) a specific cache entry.

    Args:
        key_data: Key of the cache entry to invalidate

    """
    ...

clear async

clear()

Clear all entries from the cache.

Source code in src/core/models/protocols.py

async def clear(self) -> None:
    """Clear all entries from the cache."""
    ...

load_cache async

load_cache()

Load persistent cache data from disk.

Source code in src/core/models/protocols.py

async def load_cache(self) -> None:
    """Load persistent cache data from disk."""
    ...

save_cache async

save_cache()

Save cache data to disk for persistence.

Source code in src/core/models/protocols.py

async def save_cache(self) -> None:
    """Save cache data to disk for persistence."""
    ...

get_last_run_timestamp async

get_last_run_timestamp()

Get the timestamp of the last cache run.

Returns:

Type	Description
datetime	DateTime of last run, or epoch if never run

Source code in src/core/models/protocols.py

async def get_last_run_timestamp(self) -> datetime:
    """Get the timestamp of the last cache run.

    Returns:
        DateTime of last run, or epoch if never run

    """
    ...

get_album_year_from_cache async

get_album_year_from_cache(artist, album)

Get cached album year for an artist/album pair.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
str | None	Cached year string or None if not found

Source code in src/core/models/protocols.py

async def get_album_year_from_cache(self, artist: str, album: str) -> str | None:
    """Get cached album year for an artist/album pair.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Cached year string or None if not found

    """
    ...

get_album_year_entry_from_cache async

get_album_year_entry_from_cache(artist, album)

Get full album cache entry for an artist/album pair.

Use this method when you need to check confidence level before trusting cached data.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
AlbumCacheEntry | None	Full AlbumCacheEntry or None if not found

Source code in src/core/models/protocols.py

async def get_album_year_entry_from_cache(self, artist: str, album: str) -> AlbumCacheEntry | None:
    """Get full album cache entry for an artist/album pair.

    Use this method when you need to check confidence level before trusting cached data.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Full AlbumCacheEntry or None if not found

    """
    ...

store_album_year_in_cache async

store_album_year_in_cache(
    artist, album, year, confidence=0
)

Store album year in persistent cache.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
year	str	Year to cache	required
confidence	int	Confidence score 0-100 (higher = more trustworthy)	0

Source code in src/core/models/protocols.py

async def store_album_year_in_cache(
    self,
    artist: str,
    album: str,
    year: str,
    confidence: int = 0,
) -> None:
    """Store album year in persistent cache.

    Args:
        artist: Artist name
        album: Album name
        year: Year to cache
        confidence: Confidence score 0-100 (higher = more trustworthy)

    """
    ...

invalidate_album_cache async

invalidate_album_cache(artist, album)

Invalidate cached data for a specific album.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Source code in src/core/models/protocols.py

async def invalidate_album_cache(self, artist: str, album: str) -> None:
    """Invalidate cached data for a specific album.

    Args:
        artist: Artist name
        album: Album name

    """
    ...

invalidate_all_albums async

invalidate_all_albums()

Invalidate all album cache entries.

Source code in src/core/models/protocols.py

async def invalidate_all_albums(self) -> None:
    """Invalidate all album cache entries."""
    ...

sync_cache async

sync_cache()

Synchronize cache to persistent storage.

Source code in src/core/models/protocols.py

async def sync_cache(self) -> None:
    """Synchronize cache to persistent storage."""
    ...

get_cached_api_result async

get_cached_api_result(artist, album, source)

Get cached API result for an artist/album from a specific source.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
source	str	API source identifier	required

Returns:

Type	Description
CachedApiResult | None	Cached API result or None if not found

Source code in src/core/models/protocols.py

async def get_cached_api_result(
    self,
    artist: str,
    album: str,
    source: str,
) -> CachedApiResult | None:
    """Get cached API result for an artist/album from a specific source.

    Args:
        artist: Artist name
        album: Album name
        source: API source identifier

    Returns:
        Cached API result or None if not found

    """
    ...

set_cached_api_result async

set_cached_api_result(
    artist,
    album,
    source,
    year,
    *,
    metadata=None,
    is_negative=False
)

Cache an API result for an artist/album from a specific source.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
source	str	API source identifier	required
year	str | None	Year to cache (None for negative cache)	required
metadata	dict[str, Any] | None	Optional metadata about the result	None
is_negative	bool	Whether this is a negative cache entry	False

Source code in src/core/models/protocols.py

async def set_cached_api_result(
    self,
    artist: str,
    album: str,
    source: str,
    year: str | None,
    *,
    metadata: dict[str, Any] | None = None,
    is_negative: bool = False,
) -> None:
    """Cache an API result for an artist/album from a specific source.

    Args:
        artist: Artist name
        album: Album name
        source: API source identifier
        year: Year to cache (None for negative cache)
        metadata: Optional metadata about the result
        is_negative: Whether this is a negative cache entry

    """
    ...

invalidate_for_track async

invalidate_for_track(track)

Invalidate caches impacted by a specific track update.

Source code in src/core/models/protocols.py

async def invalidate_for_track(self, track: TrackDict) -> None:
    """Invalidate caches impacted by a specific track update."""
    ...

generate_album_key staticmethod

generate_album_key(artist, album)

Generate a unique key for an artist/album pair.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
str	Unique key string (typically a hash)

Source code in src/core/models/protocols.py

@staticmethod
def generate_album_key(artist: str, album: str) -> str:
    """Generate a unique key for an artist/album pair.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Unique key string (typically a hash)

    """
    ...

ExternalApiServiceProtocol

Bases: Protocol

Protocol defining the interface for external API services.

This protocol defines methods for fetching music metadata from external sources like MusicBrainz, Last.fm, and Discogs.

initialize async

initialize(force=False)

Initialize the external API service.

Parameters:

Name	Type	Description	Default
force	bool	Force re-initialization even if already initialized	False

Source code in src/core/models/protocols.py

async def initialize(self, force: bool = False) -> None:
    """Initialize the external API service.

    Args:
        force: Force re-initialization even if already initialized

    """
    ...

close async

close()

Close all connections and clean up resources.

Source code in src/core/models/protocols.py

async def close(self) -> None:
    """Close all connections and clean up resources."""
    ...

get_album_year async

get_album_year(
    artist,
    album,
    current_library_year=None,
    earliest_track_added_year=None,
)

Determine the original release year for an album using optimized API calls and revised scoring.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
current_library_year	str | None	Current year in library (optional)	None
earliest_track_added_year	int | None	Earliest year any track was added to library (optional)	None

Returns:

Type	Description
str | None	Tuple of (year_string, is_definitive, confidence_score, year_scores) where:
bool	year_string: Best determined year or None
int	is_definitive: True if high confidence in the result
dict[str, int]	confidence_score: 0-100 confidence percentage
tuple[str | None, bool, int, dict[str, int]]	year_scores: Dict mapping each year to its max score from APIs

Source code in src/core/models/protocols.py

async def get_album_year(
    self,
    artist: str,
    album: str,
    current_library_year: str | None = None,
    earliest_track_added_year: int | None = None,
) -> tuple[str | None, bool, int, dict[str, int]]:
    """Determine the original release year for an album using optimized API calls and revised scoring.

    Args:
        artist: Artist name
        album: Album name
        current_library_year: Current year in library (optional)
        earliest_track_added_year: Earliest year any track was added to library (optional)

    Returns:
        Tuple of (year_string, is_definitive, confidence_score, year_scores) where:
        - year_string: Best determined year or None
        - is_definitive: True if high confidence in the result
        - confidence_score: 0-100 confidence percentage
        - year_scores: Dict mapping each year to its max score from APIs

    """
    ...

get_artist_activity_period async

get_artist_activity_period(artist_norm)

Retrieve the period of activity for an artist from MusicBrainz, with caching.

Parameters:

Name	Type	Description	Default
artist_norm	str	Normalized artist name	required

Returns:

Type	Description
tuple[int | None, int | None]	Tuple of (start_year, end_year) as integers or (None, None) if not found

Source code in src/core/models/protocols.py

async def get_artist_activity_period(
    self,
    artist_norm: str,
) -> tuple[int | None, int | None]:
    """Retrieve the period of activity for an artist from MusicBrainz, with caching.

    Args:
        artist_norm: Normalized artist name

    Returns:
        Tuple of (start_year, end_year) as integers or (None, None) if not found

    """
    ...

get_artist_start_year async

get_artist_start_year(artist_norm)

Get artist's career start year with caching and fallback.

Uses MusicBrainz as primary source, iTunes as fallback. Results are cached for performance.

Parameters:

Name	Type	Description	Default
artist_norm	str	Normalized artist name	required

Returns:

Type	Description
int | None	Artist's career start year, or None if not found

Source code in src/core/models/protocols.py

async def get_artist_start_year(
    self,
    artist_norm: str,
) -> int | None:
    """Get artist's career start year with caching and fallback.

    Uses MusicBrainz as primary source, iTunes as fallback.
    Results are cached for performance.

    Args:
        artist_norm: Normalized artist name

    Returns:
        Artist's career start year, or None if not found

    """
    ...

get_year_from_discogs async

get_year_from_discogs(artist, album)

Fetch the earliest release year for an album from Discogs.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
str | None	Year string or None if not found

Source code in src/core/models/protocols.py

async def get_year_from_discogs(
    self,
    artist: str,
    album: str,
) -> str | None:
    """Fetch the earliest release year for an album from Discogs.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Year string or None if not found

    """
    ...

PendingVerificationServiceProtocol

Bases: Protocol

Protocol defining the interface for pending verification services.

This protocol defines methods for managing albums that need manual verification or have pending updates.

initialize async

initialize()

Initialize the pending verification service.

Source code in src/core/models/protocols.py

async def initialize(self) -> None:
    """Initialize the pending verification service."""
    ...

mark_for_verification async

mark_for_verification(
    artist,
    album,
    reason="no_year_found",
    metadata=None,
    recheck_days=None,
)

Mark an album for future verification.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
reason	str	Reason for marking for verification	'no_year_found'
metadata	dict[str, Any] | None	Additional metadata to store	None
recheck_days	int | None	Optional override for verification interval in days	None

Source code in src/core/models/protocols.py

async def mark_for_verification(
    self,
    artist: str,
    album: str,
    reason: str = "no_year_found",
    metadata: dict[str, Any] | None = None,
    recheck_days: int | None = None,
) -> None:
    """Mark an album for future verification.

    Args:
        artist: Artist name
        album: Album name
        reason: Reason for marking for verification
        metadata: Additional metadata to store
        recheck_days: Optional override for verification interval in days

    """
    ...

remove_from_pending async

remove_from_pending(artist, album)

Remove an album from pending verification.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Source code in src/core/models/protocols.py

async def remove_from_pending(
    self,
    artist: str,
    album: str,
) -> None:
    """Remove an album from pending verification.

    Args:
        artist: Artist name
        album: Album name

    """
    ...

get_entry async

get_entry(artist, album)

Get pending entry for artist/album if exists.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
PendingAlbumEntry | None	PendingAlbumEntry if found, None otherwise.

Source code in src/core/models/protocols.py

async def get_entry(
    self,
    artist: str,
    album: str,
) -> PendingAlbumEntry | None:
    """Get pending entry for artist/album if exists.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        PendingAlbumEntry if found, None otherwise.

    """
    ...

get_attempt_count async

get_attempt_count(artist, album)

Get current verification attempt count for an album.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
int	Number of verification attempts made (0 if not in pending list).

Source code in src/core/models/protocols.py

async def get_attempt_count(
    self,
    artist: str,
    album: str,
) -> int:
    """Get current verification attempt count for an album.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Number of verification attempts made (0 if not in pending list).

    """
    ...

is_verification_needed async

is_verification_needed(artist, album)

Check if an album needs verification now.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required

Returns:

Type	Description
bool	True if the verification period has elapsed, False otherwise.

Source code in src/core/models/protocols.py

async def is_verification_needed(
    self,
    artist: str,
    album: str,
) -> bool:
    """Check if an album needs verification now.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        True if the verification period has elapsed, False otherwise.

    """
    ...

get_all_pending_albums async

get_all_pending_albums()

Get all pending albums.

Returns:

Type	Description
list[PendingAlbumEntry]	List of PendingAlbumEntry objects

Source code in src/core/models/protocols.py

async def get_all_pending_albums(
    self,
) -> list[PendingAlbumEntry]:
    """Get all pending albums.

    Returns:
        List of PendingAlbumEntry objects

    """
    ...

generate_problematic_albums_report async

generate_problematic_albums_report(min_attempts=3)

Generate report of albums that failed to get year after multiple attempts.

Parameters:

Name	Type	Description	Default
min_attempts	int	Minimum number of attempts to consider problematic	3

Returns:

Type	Description
int	Number of problematic albums found

Source code in src/core/models/protocols.py

async def generate_problematic_albums_report(
    self,
    min_attempts: int = 3,
) -> int:
    """Generate report of albums that failed to get year after multiple attempts.

    Args:
        min_attempts: Minimum number of attempts to consider problematic

    Returns:
        Number of problematic albums found

    """
    ...

should_auto_verify async

should_auto_verify()

Check if automatic pending verification should run.

Returns True if: - auto_verify_days has passed since last verification - No previous verification exists - auto_verify_days > 0 (feature enabled)

Returns:

Type	Description
bool	True if auto-verify should run, False otherwise.

Source code in src/core/models/protocols.py

async def should_auto_verify(self) -> bool:
    """Check if automatic pending verification should run.

    Returns True if:
    - auto_verify_days has passed since last verification
    - No previous verification exists
    - auto_verify_days > 0 (feature enabled)

    Returns:
        True if auto-verify should run, False otherwise.

    """
    ...

update_verification_timestamp async

update_verification_timestamp()

Update the last pending verification timestamp file.

Source code in src/core/models/protocols.py

async def update_verification_timestamp(self) -> None:
    """Update the last pending verification timestamp file."""
    ...

RateLimiterProtocol

Bases: Protocol

Protocol defining the interface for rate limiting services.

This protocol ensures API calls respect rate limits and prevent overwhelming external services.

acquire async

acquire()

Acquire permission to make a request.

Returns:

Type	Description
float	Wait time before the request was allowed

Source code in src/core/models/protocols.py

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

    Returns:
        Wait time before the request was allowed

    """
    ...

release

release()

Release a request slot (for cleanup/error cases).

Source code in src/core/models/protocols.py

def release(self) -> None:
    """Release a request slot (for cleanup/error cases)."""
    ...

get_stats

get_stats()

Get rate limiter statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary containing rate limit statistics

Source code in src/core/models/protocols.py

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

    Returns:
        Dictionary containing rate limit statistics

    """
    ...