update_executor

Track update execution module.

This module handles track update operations including: - Single and batch property updates - Dry run recording - Security validation - Cache invalidation after updates

TrackUpdateExecutor

TrackUpdateExecutor(
    ap_client,
    cache_service,
    security_validator,
    config,
    console_logger,
    error_logger,
    analytics,
    *,
    dry_run=False
)

Executes track metadata updates via AppleScript.

This class handles: - Individual and batch property updates - Security validation of parameters - Dry run mode recording - Cache invalidation after successful updates

Initialize the update executor.

Parameters:

Name	Type	Description	Default
ap_client	AppleScriptClientProtocol	AppleScript client for executing updates	required
cache_service	CacheServiceProtocol	Cache service for invalidation	required
security_validator	SecurityValidator	Validator for sanitizing inputs	required
config	AppConfig	Typed application configuration	required
console_logger	Logger	Logger for info/debug messages	required
error_logger	Logger	Logger for error messages	required
analytics	AnalyticsProtocol	Service for performance tracking	required
dry_run	bool	If True, record actions without executing	False

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

def __init__(
    self,
    ap_client: AppleScriptClientProtocol,
    cache_service: CacheServiceProtocol,
    security_validator: SecurityValidator,
    config: AppConfig,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    analytics: AnalyticsProtocol,
    *,
    dry_run: bool = False,
) -> None:
    """Initialize the update executor.

    Args:
        ap_client: AppleScript client for executing updates
        cache_service: Cache service for invalidation
        security_validator: Validator for sanitizing inputs
        config: Typed application configuration
        console_logger: Logger for info/debug messages
        error_logger: Logger for error messages
        analytics: Service for performance tracking
        dry_run: If True, record actions without executing
    """
    self.ap_client = ap_client
    self.cache_service = cache_service
    self.security_validator = security_validator
    self.config = config
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.analytics = analytics
    self.dry_run = dry_run
    self._dry_run_actions: list[dict[str, Any]] = []

set_dry_run

set_dry_run(dry_run)

Update dry run mode.

Parameters:

Name	Type	Description	Default
dry_run	bool	New dry run state	required

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

def set_dry_run(self, dry_run: bool) -> None:
    """Update dry run mode.

    Args:
        dry_run: New dry run state
    """
    self.dry_run = dry_run
    if not dry_run:
        self._dry_run_actions.clear()

get_dry_run_actions

get_dry_run_actions()

Get the list of dry-run actions recorded.

Returns:

Type	Description
list[dict[str, Any]]	List of dry-run action dictionaries

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

def get_dry_run_actions(self) -> list[dict[str, Any]]:
    """Get the list of dry-run actions recorded.

    Returns:
        List of dry-run action dictionaries
    """
    return self._dry_run_actions

update_track_async async

update_track_async(
    track_id,
    new_track_name=None,
    new_album_name=None,
    new_genre=None,
    new_year=None,
    track_status=None,
    original_artist=None,
    original_album=None,
    original_track=None,
)

Update multiple properties of a track.

Parameters:

Name	Type	Description	Default
track_id	str	ID of the track to update	required
new_track_name	str | None	New track name (optional)	None
new_album_name	str | None	New album name (optional)	None
new_genre	str | None	New genre (optional)	None
new_year	str | None	New year (optional)	None
track_status	str | None	Track status to check for prerelease (optional)	None
original_artist	str | None	Original artist name for contextual logging (optional)	None
original_album	str | None	Original album name for contextual logging (optional)	None
original_track	str | None	Original track name for contextual logging (optional)	None

Returns:

Type	Description
bool	True if all updates are successful, False if any failed

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

@track_instance_method("track_update")
async def update_track_async(
    self,
    track_id: str,
    new_track_name: str | None = None,
    new_album_name: str | None = None,
    new_genre: str | None = None,
    new_year: str | None = None,
    track_status: str | None = None,
    original_artist: str | None = None,
    original_album: str | None = None,
    original_track: str | None = None,
) -> bool:
    """Update multiple properties of a track.

    Args:
        track_id: ID of the track to update
        new_track_name: New track name (optional)
        new_album_name: New album name (optional)
        new_genre: New genre (optional)
        new_year: New year (optional)
        track_status: Track status to check for prerelease (optional)
        original_artist: Original artist name for contextual logging (optional)
        original_album: Original album name for contextual logging (optional)
        original_track: Original track name for contextual logging (optional)

    Returns:
        True if all updates are successful, False if any failed

    """
    # Check if the track is prerelease (read-only) - prevent update attempts
    if self._is_read_only_track(track_status, track_id):
        return False

    # Validate and sanitize all input parameters
    validated_params = self._validate_and_sanitize_update_parameters(track_id, new_track_name, new_album_name, new_genre, new_year)
    if validated_params is None:
        return False

    (
        sanitized_track_id,
        sanitized_track_name,
        sanitized_album_name,
        sanitized_genre,
        sanitized_year,
    ) = validated_params

    # Handle dry run mode
    if self.dry_run:
        return self._handle_dry_run_update(
            sanitized_track_id,
            sanitized_track_name,
            sanitized_album_name,
            sanitized_genre,
            sanitized_year,
        )

    # Perform actual updates
    return await self._perform_property_updates(
        sanitized_track_id,
        sanitized_track_name,
        sanitized_album_name,
        sanitized_genre,
        sanitized_year,
        original_artist,
        original_album,
        original_track,
    )

update_artist_async async

update_artist_async(
    track,
    new_artist_name,
    *,
    original_artist=None,
    update_album_artist=False
)

Update the artist name for a track.

Parameters:

Name	Type	Description	Default
track	TrackDict	Track dictionary representing the target track	required
new_artist_name	str	Artist name to apply	required
original_artist	str | None	Original artist for logging context (optional)	None
update_album_artist	bool	If True, also update album_artist field when it matches the old or new artist name from configuration mapping	False

Returns:

Type	Description
bool	True if update succeeded, False otherwise

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

@track_instance_method("track_artist_update")
async def update_artist_async(
    self,
    track: TrackDict,
    new_artist_name: str,
    *,
    original_artist: str | None = None,
    update_album_artist: bool = False,
) -> bool:
    """Update the artist name for a track.

    Args:
        track: Track dictionary representing the target track
        new_artist_name: Artist name to apply
        original_artist: Original artist for logging context (optional)
        update_album_artist: If True, also update album_artist field when it matches
                             the old or new artist name from configuration mapping

    Returns:
        True if update succeeded, False otherwise
    """
    prepared = self._prepare_artist_update(track, new_artist_name)
    if prepared is None:
        return False

    sanitized_track_id, sanitized_artist, current_artist = prepared

    if self.dry_run:
        return self._handle_dry_run_update(
            sanitized_track_id,
            None,
            None,
            None,
            None,
            sanitized_artist_name=sanitized_artist,
        )

    # Prepare updates list: always update artist
    updates: list[tuple[str, str]] = [("artist", sanitized_artist)]

    # Update album_artist only if explicitly requested (e.g., from artist rename configuration)
    # This ensures we only sync album_artist when both fields are from the same mapping
    if update_album_artist:
        album_artist = getattr(track, "album_artist", None)
        if isinstance(album_artist, str):
            normalized_album_artist = album_artist.strip()
            # Check if album_artist matches either old or new name from configuration
            if normalized_album_artist in (current_artist, sanitized_artist):
                updates.append(("album_artist", sanitized_artist))

    success = await self._apply_track_updates(
        sanitized_track_id,
        updates,
        original_artist=original_artist or current_artist,
        album=track.album,
        track=track.name,
    )

    if success:
        track.artist = sanitized_artist
        # Update album_artist in TrackDict if it was updated in Music.app
        if len(updates) > 1:
            track.album_artist = sanitized_artist

    return success