cache_config

Smart Cache Configuration with Event-Driven Invalidation.

This module provides intelligent cache configuration that understands the lifecycle of media library tracks and optimizes caching strategies accordingly.

Key Features: - Event-driven invalidation for media library tracks - Content-type aware TTL management - Integration with fingerprint-based change detection - Lifecycle management for persistent vs temporary cache entries

CacheContentType

Bases: Enum

Types of cached content with different lifecycle requirements.

InvalidationStrategy

Bases: Enum

Cache invalidation strategies.

CachePolicy dataclass

CachePolicy(
    content_type,
    ttl_seconds,
    invalidation_strategy,
    max_size_mb=None,
    cleanup_threshold=0.8,
    description="",
)

Configuration for a specific type of cached content.

SmartCacheConfig

SmartCacheConfig(config=None)

Smart cache configuration with content-aware policies.

Initialize smart cache configuration.

Parameters:

Name	Type	Description	Default
config	AppConfig | None	Optional typed application configuration	None

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

def __init__(self, config: AppConfig | None = None) -> None:
    """Initialize smart cache configuration.

    Args:
        config: Optional typed application configuration
    """
    self.logger = logging.getLogger(__name__)
    self._config = config
    self._policies = self._create_default_policies()

get_policy

get_policy(content_type)

Get cache policy for specific content type.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required

Returns:

Type	Description
CachePolicy	Cache policy with TTL and invalidation strategy

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

def get_policy(self, content_type: CacheContentType) -> CachePolicy:
    """Get cache policy for specific content type.

    Args:
        content_type: Type of cached content

    Returns:
        Cache policy with TTL and invalidation strategy
    """
    return self._policies[content_type]

get_ttl

get_ttl(content_type)

Get TTL in seconds for specific content type.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required

Returns:

Type	Description
int	TTL in seconds

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

def get_ttl(self, content_type: CacheContentType) -> int:
    """Get TTL in seconds for specific content type.

    Args:
        content_type: Type of cached content

    Returns:
        TTL in seconds
    """
    return self._policies[content_type].ttl_seconds

should_use_fingerprint_validation

should_use_fingerprint_validation(content_type)

Check if content type should use fingerprint-based validation.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required

Returns:

Type	Description
bool	True if fingerprint validation should be used

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

def should_use_fingerprint_validation(self, content_type: CacheContentType) -> bool:
    """Check if content type should use fingerprint-based validation.

    Args:
        content_type: Type of cached content

    Returns:
        True if fingerprint validation should be used
    """
    policy = self._policies[content_type]
    return policy.invalidation_strategy in [InvalidationStrategy.FINGERPRINT_BASED, InvalidationStrategy.HYBRID]

should_use_event_invalidation

should_use_event_invalidation(content_type)

Check if content type should use event-driven invalidation.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required

Returns:

Type	Description
bool	True if event-driven invalidation should be used

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

def should_use_event_invalidation(self, content_type: CacheContentType) -> bool:
    """Check if content type should use event-driven invalidation.

    Args:
        content_type: Type of cached content

    Returns:
        True if event-driven invalidation should be used
    """
    policy = self._policies[content_type]
    return policy.invalidation_strategy in [InvalidationStrategy.EVENT_DRIVEN, InvalidationStrategy.HYBRID]

is_persistent_cache staticmethod

is_persistent_cache(content_type)

Check if content type represents persistent cache data.

Persistent cache should survive application restarts and only be invalidated by specific events.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required

Returns:

Type	Description
bool	True if cache should be persistent

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

@staticmethod
def is_persistent_cache(content_type: CacheContentType) -> bool:
    """Check if content type represents persistent cache data.

    Persistent cache should survive application restarts and
    only be invalidated by specific events.

    Args:
        content_type: Type of cached content

    Returns:
        True if cache should be persistent
    """
    return content_type in [CacheContentType.TRACK_METADATA, CacheContentType.SUCCESSFUL_API_METADATA, CacheContentType.ALBUM_YEAR]

CacheEventType

Bases: Enum

Types of cache invalidation events.

CacheEvent dataclass

CacheEvent(
    event_type,
    track_id=None,
    fingerprint=None,
    metadata=None,
)

Cache event for event-driven invalidation.

__post_init__

__post_init__()

Validate event data.

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

def __post_init__(self) -> None:
    """Validate event data."""
    if self.event_type in [CacheEventType.TRACK_ADDED, CacheEventType.TRACK_REMOVED, CacheEventType.TRACK_MODIFIED] and not self.track_id:
        msg = f"track_id required for {self.event_type.value}"
        raise ValueError(msg)
    if self.event_type == CacheEventType.FINGERPRINT_CHANGED and not self.fingerprint:
        msg = "fingerprint required for fingerprint_changed event"
        raise ValueError(msg)

EventDrivenCacheManager

EventDrivenCacheManager(config)

Manages event-driven cache invalidation.

Initialize event-driven cache manager.

Parameters:

Name	Type	Description	Default
config	SmartCacheConfig	Smart cache configuration instance	required

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

def __init__(self, config: SmartCacheConfig) -> None:
    """Initialize event-driven cache manager.

    Args:
        config: Smart cache configuration instance
    """
    self.config = config
    self.logger = logging.getLogger(__name__)
    self._event_handlers: dict[CacheEventType, list[Callable[[CacheEvent], None]]] = {}

register_event_handler

register_event_handler(event_type, handler)

Register event handler for specific event type.

Parameters:

Name	Type	Description	Default
event_type	CacheEventType	Type of cache event	required
handler	Callable[[CacheEvent], None]	Callable to handle the event	required

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

def register_event_handler(self, event_type: CacheEventType, handler: Callable[[CacheEvent], None]) -> None:
    """Register event handler for specific event type.

    Args:
        event_type: Type of cache event
        handler: Callable to handle the event
    """
    if event_type not in self._event_handlers:
        self._event_handlers[event_type] = []

    self._event_handlers[event_type].append(handler)
    self.logger.debug("Registered handler for %s", event_type.value)

emit_event

emit_event(event)

Emit cache event to trigger invalidation.

Parameters:

Name	Type	Description	Default
event	CacheEvent	Cache event to emit	required

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

def emit_event(self, event: CacheEvent) -> None:
    """Emit cache event to trigger invalidation.

    Args:
        event: Cache event to emit
    """
    self.logger.info("Cache event: %s %s", event.event_type.value, event.track_id or "")

    handlers = self._event_handlers.get(event.event_type, [])
    for handler in handlers:
        try:
            handler(event)
        except (RuntimeError, ValueError, TypeError, AttributeError, KeyError) as exc:
            # Catch common runtime errors; ensures remaining handlers execute
            self.logger.exception("Event handler %r failed: %s", handler, exc)

should_invalidate_for_event

should_invalidate_for_event(content_type, event)

Check if cache should be invalidated for given event.

Parameters:

Name	Type	Description	Default
content_type	CacheContentType	Type of cached content	required
event	CacheEvent	Cache event that occurred	required

Returns:

Type	Description
bool	True if cache should be invalidated

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

def should_invalidate_for_event(self, content_type: CacheContentType, event: CacheEvent) -> bool:
    """Check if cache should be invalidated for given event.

    Args:
        content_type: Type of cached content
        event: Cache event that occurred

    Returns:
        True if cache should be invalidated
    """
    # Manual invalidation affects all content types that support events
    if event.event_type == CacheEventType.MANUAL_INVALIDATION:
        return self.config.should_use_event_invalidation(content_type)

    if not self.config.should_use_event_invalidation(content_type):
        return False

    # Track metadata cache should invalidate for track events
    if content_type == CacheContentType.TRACK_METADATA:
        return event.event_type in [CacheEventType.TRACK_REMOVED, CacheEventType.TRACK_MODIFIED, CacheEventType.FINGERPRINT_CHANGED]

    # API metadata cache should invalidate when tracks are removed/modified
    if content_type == CacheContentType.SUCCESSFUL_API_METADATA:
        return event.event_type in [
            CacheEventType.TRACK_REMOVED,  # Track removed → invalidate API data
            CacheEventType.TRACK_MODIFIED,  # Track metadata changed → may need refresh
        ]

    # Library sync may affect album years
    return content_type == CacheContentType.ALBUM_YEAR and event.event_type == CacheEventType.LIBRARY_SYNC