Skip to content

orchestrator

Main orchestrator module for Music Genre Updater.

This module handles the high-level coordination of all operations.

Orchestrator 

Orchestrator(deps)

Orchestrates the entire music update workflow.

Initialize the orchestrator with dependencies.

Parameters:

Name Type Description Default
deps DependencyContainer

Dependency container with all required services

 required
Source code in src/app/orchestrator.py 
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, deps: DependencyContainer) -> None:
    """Initialize the orchestrator with dependencies.

    Args:
        deps: Dependency container with all required services

    """
    self.deps = deps
    self.music_updater = MusicUpdater(deps)
    self.config = deps.app_config
    self.console_logger = deps.console_logger
    self.error_logger = deps.error_logger

run_command async 

run_command(args)

Execute the appropriate command based on arguments.

Parameters:

Name Type Description Default
args Namespace

Parsed command-line arguments from argparse.

 required
Source code in src/app/orchestrator.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def run_command(self, args: argparse.Namespace) -> None:
    """Execute the appropriate command based on arguments.

    Args:
        args: Parsed command-line arguments from argparse.

    """
    # Reset per-run state
    reset_cleaning_exceptions_log()

    # Handle --fresh flag: clear all caches and snapshots
    if getattr(args, "fresh", False):
        self.console_logger.info("--fresh flag set: clearing all caches and snapshots...")
        await self.deps.cache_service.clear()
        self.deps.library_snapshot_service.clear_snapshot()
        self.console_logger.info("All caches cleared. Will fetch fresh data from Music.app.")

    command = getattr(args, "command", None)

    # Check if Music app is running for commands that depend on it
    if self._requires_music_app(command) and not is_music_app_running(self.error_logger):
        self.console_logger.error(
            "Music.app is not running - cannot execute '%s' command. Please start Music.app before running this script.",
            command or "default",
        )
        return

    # Set dry-run context if needed
    if args.dry_run or getattr(args, "test_mode", False):
        test_artists = set(self.config.development.test_artists)
        mode = "test" if getattr(args, "test_mode", False) else "dry_run"
        self.music_updater.set_dry_run_context(mode, test_artists)

    # ALWAYS apply test_artists from config if configured (even in --force mode)
    elif test_artists_config := set(self.config.development.test_artists):
        self.console_logger.info("Using test_artists from config in normal mode: %s", sorted(test_artists_config))
        self.music_updater.set_dry_run_context("normal", test_artists_config)

    # Route to an appropriate command
    match args.command:
        case "clean_artist" | "clean":
            await self._run_clean_artist(args)
        case "update_years" | "years":
            await self._run_update_years(args)
        case "update_genres" | "genres":
            await self._run_update_genres(args)
        case "revert_years" | "revert":
            await self._run_revert_years(args)
        case "restore_release_years" | "restore":
            await self._run_restore_release_years(args)
        case "verify_database" | "verify-db":
            await self._run_verify_database(args)
        case "verify_pending" | "pending":
            await self._run_verify_pending(args)
        case "batch":
            await self._run_batch(args)
        case "rotate_keys" | "rotate-keys":
            self._run_rotate_encryption_keys(args)
        case _:
            await self._run_main_workflow(args)