album_cache

Album cache service with TTL-aware entries and CSV persistence.

AlbumCacheService

AlbumCacheService(config, logger=None)

Specialized cache service for album release years with CSV persistence.

Initialize album cache service.

Parameters:

Name	Type	Description	Default
config	AppConfig	Typed application configuration	required
logger	Logger | None	Optional logger instance	None

Source code in src/services/cache/album_cache.py

def __init__(self, config: AppConfig, logger: logging.Logger | None = None) -> None:
    """Initialize album cache service.

    Args:
        config: Typed application configuration
        logger: Optional logger instance
    """
    self.config = config
    self.logger = logger or logging.getLogger(__name__)
    self.cache_config = SmartCacheConfig(config)
    self.policy = self.cache_config.get_policy(CacheContentType.ALBUM_YEAR)

    # Album years cache: {hash_key: AlbumCacheEntry}
    self.album_years_cache: dict[str, AlbumCacheEntry] = {}

    # Lock for thread-safe cache operations
    self._cache_lock = asyncio.Lock()

    # Cache file paths - use get_full_log_path to ensure proper logs_base_dir integration
    self.album_years_cache_file = Path(get_full_log_path(config, "album_years_cache_file", "cache/album_years.csv"))

initialize async

initialize()

Initialize album cache by loading data from disk.

Source code in src/services/cache/album_cache.py

async def initialize(self) -> None:
    """Initialize album cache by loading data from disk."""
    self.logger.info("Initializing %s...", LogFormat.entity("AlbumCacheService"))
    await self._load_album_years_cache()
    self.logger.info("%s initialized with %d albums", LogFormat.entity("AlbumCacheService"), len(self.album_years_cache))

get_album_year async

get_album_year(artist, album)

Get album release year from cache.

Parameters:

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

Returns:

Type	Description
str | None	Album release year if found, None otherwise

Source code in src/services/cache/album_cache.py

async def get_album_year(self, artist: str, album: str) -> str | None:
    """Get album release year from cache.

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Album release year if found, None otherwise
    """
    async with self._cache_lock:
        key = UnifiedHashService.hash_album_key(artist, album)

        if key not in self.album_years_cache:
            self.logger.debug("Album year cache miss: %s - %s", artist, album)
            return None

        entry = self.album_years_cache[key]

        # Validate cache entry consistency (detect true hash collision)
        artist_mismatch = not are_names_equal(entry.artist, artist)
        album_mismatch = not are_names_equal(entry.album, album)
        if artist_mismatch or album_mismatch:
            self.logger.warning(
                "Hash collision detected: requested '%s - %s', found '%s - %s'",
                artist,
                album,
                entry.artist,
                entry.album,
            )
            # Don't delete - keep the original entry, just miss for this request
            return None

        if self._is_entry_expired(entry):
            self.logger.debug("Album year cache expired: %s - %s", artist, album)
            del self.album_years_cache[key]
            return None

        self.logger.debug("Album year cache hit: %s - %s = %s", artist, album, entry.year)
        return entry.year

get_album_year_entry async

get_album_year_entry(artist, album)

Get full album cache entry (not just year).

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 if found and not expired, None otherwise

Source code in src/services/cache/album_cache.py

async def get_album_year_entry(self, artist: str, album: str) -> AlbumCacheEntry | None:
    """Get full album cache entry (not just year).

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

    Args:
        artist: Artist name
        album: Album name

    Returns:
        Full AlbumCacheEntry if found and not expired, None otherwise
    """
    async with self._cache_lock:
        key = UnifiedHashService.hash_album_key(artist, album)

        if key not in self.album_years_cache:
            self.logger.debug("Album year cache miss: %s - %s", artist, album)
            return None

        entry = self.album_years_cache[key]

        # Validate cache entry consistency (detect true hash collision)
        artist_mismatch = not are_names_equal(entry.artist, artist)
        album_mismatch = not are_names_equal(entry.album, album)
        if artist_mismatch or album_mismatch:
            self.logger.warning(
                "Hash collision detected: requested '%s - %s', found '%s - %s'",
                artist,
                album,
                entry.artist,
                entry.album,
            )
            return None

        if self._is_entry_expired(entry):
            self.logger.debug("Album year cache expired: %s - %s", artist, album)
            del self.album_years_cache[key]
            return None

        self.logger.debug(
            "Album year cache entry: %s - %s = %s (confidence %d%%)",
            artist,
            album,
            entry.year,
            entry.confidence,
        )
        return entry

store_album_year async

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

Store album release year in cache.

Parameters:

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

Source code in src/services/cache/album_cache.py

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

    Args:
        artist: Artist name
        album: Album name
        year: Album release year
        confidence: Confidence score 0-100 (higher = more trustworthy)
    """
    async with self._cache_lock:
        key = UnifiedHashService.hash_album_key(artist, album)

        # Store with normalized values for consistency
        normalized_artist = artist.strip()
        normalized_album = album.strip()
        normalized_year = year.strip()

        self.album_years_cache[key] = AlbumCacheEntry(
            artist=normalized_artist,
            album=normalized_album,
            year=normalized_year,
            timestamp=time.time(),
            confidence=confidence,
        )
        self.logger.debug(
            "Stored album year: %s - %s = %s (confidence %d%%)",
            normalized_artist,
            normalized_album,
            normalized_year,
            confidence,
        )

invalidate_album async

invalidate_album(artist, album)

Invalidate specific album from cache.

Parameters:

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

Source code in src/services/cache/album_cache.py

async def invalidate_album(self, artist: str, album: str) -> None:
    """Invalidate specific album from cache.

    Args:
        artist: Artist name
        album: Album name
    """
    async with self._cache_lock:
        key = UnifiedHashService.hash_album_key(artist, album)

        if key in self.album_years_cache:
            del self.album_years_cache[key]
            self.logger.info("Invalidated album cache: %s - %s", artist, album)

invalidate_all async

invalidate_all()

Clear all album cache entries.

Source code in src/services/cache/album_cache.py

async def invalidate_all(self) -> None:
    """Clear all album cache entries."""
    async with self._cache_lock:
        count = len(self.album_years_cache)
        self.album_years_cache.clear()
        self.logger.info("Cleared all album cache entries (%d items)", count)

save_to_disk async

save_to_disk()

Save album cache to CSV file.

Source code in src/services/cache/album_cache.py

async def save_to_disk(self) -> None:
    """Save album cache to CSV file."""
    if not self.album_years_cache:
        self.logger.debug("Album cache is empty, skipping save")
        return

    def blocking_save() -> None:
        """Synchronous save operation for thread executor."""
        try:
            # Ensure directory exists
            ensure_directory(str(self.album_years_cache_file.parent))

            # Prepare data for CSV
            items = list(self.album_years_cache.values())

            # Write CSV file
            self._write_csv_data(str(self.album_years_cache_file), items)
            self.logger.info("Album cache saved to [cyan]%s[/cyan] (%d entries)", self.album_years_cache_file.name, len(items))

        except (OSError, UnicodeError) as e:
            self.logger.exception("Failed to save album cache: %s", e)
            raise

    # Run in thread executor to avoid blocking
    await asyncio.get_running_loop().run_in_executor(None, blocking_save)

get_stats

get_stats()

Get album cache statistics.

Returns:

Type	Description
dict[str, Any]	Dictionary containing cache statistics

Source code in src/services/cache/album_cache.py

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

    Returns:
        Dictionary containing cache statistics
    """
    ttl_seconds = self.policy.ttl_seconds
    return {
        "total_albums": len(self.album_years_cache),
        "cache_file": str(self.album_years_cache_file),
        "cache_file_exists": self.album_years_cache_file.exists(),
        "content_type": CacheContentType.ALBUM_YEAR.value,
        "ttl_policy": ttl_seconds,
        "persistent": ttl_seconds >= self.cache_config.INFINITE_TTL,
    }