dependency_container

Dependency Injection Container Module.

Manages the lifecycle and dependency relationships between application services. Centralizes service initialization, configuration, and proper shutdown procedures. Provides access to configured service instances including loggers, analytics, API clients, and core application components. Handles circular imports and asynchronous resource management.

InitializableService

Bases: Protocol

Protocol for services that can be initialized.

Note: Services may have different initialize signatures, handled dynamically by _initialize_service method.

initialize

initialize(*args, **kwargs)

Initialize the service.

Source code in src/services/dependency_container.py

def initialize(self, *args: Any, **kwargs: Any) -> Awaitable[None]:
    """Initialize the service."""
    ...

DependencyContainer

DependencyContainer(
    config_path,
    console_logger,
    error_logger,
    analytics_logger,
    db_verify_logger,
    *,
    logging_listener=None,
    dry_run=False,
    skip_api_validation=False
)

Dependency injection container for the application.

This class manages the lifecycle and dependencies of all services in the application.

Initialize the dependency container.

Parameters:

Name	Type	Description	Default
config_path	str	Path to the configuration file	required
console_logger	Logger	Logger for console output	required
error_logger	Logger	Logger for error messages	required
analytics_logger	Logger	Logger for analytics events	required
db_verify_logger	Logger	Logger for database verification operations	required
logging_listener	SafeQueueListener | None	Optional queue listener for logging	None
dry_run	bool	Whether to run in dry-run mode (no changes made)	False
skip_api_validation	bool	Whether to skip API auth validation (for non-API commands)	False

Source code in src/services/dependency_container.py

def __init__(
    self,
    config_path: str,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    analytics_logger: logging.Logger,
    db_verify_logger: logging.Logger,
    *,
    logging_listener: SafeQueueListener | None = None,
    dry_run: bool = False,
    skip_api_validation: bool = False,
) -> None:
    """Initialize the dependency container.

    Args:
        config_path: Path to the configuration file
        console_logger: Logger for console output
        error_logger: Logger for error messages
        analytics_logger: Logger for analytics events
        db_verify_logger: Logger for database verification operations
        logging_listener: Optional queue listener for logging
        dry_run: Whether to run in dry-run mode (no changes made)
        skip_api_validation: Whether to skip API auth validation (for non-API commands)

    """
    # Initialize logger properties first
    self._console_logger = console_logger
    self._error_logger = error_logger
    self._analytics_logger = analytics_logger
    self._db_verify_logger = db_verify_logger
    self._listener = logging_listener

    # Initialize service references
    self._config_path = config_path
    self._app_config: AppConfig | None = None
    self._analytics: Analytics | None = None
    self._ap_client: AppleScriptClientProtocol | None = None
    self._cache_service: CacheOrchestrator | None = None
    self._library_snapshot_service: LibrarySnapshotService | None = None
    self._pending_verification_service: PendingVerificationService | None = None
    self._api_orchestrator: ExternalApiOrchestrator | None = None
    self._retry_handler: DatabaseRetryHandler | None = None
    self._dry_run = dry_run
    self._skip_api_validation = skip_api_validation

dry_run property

dry_run

Whether the application runs in dry-run mode.

app_config property

app_config

Validated application configuration; raises if not yet loaded.

config_path property

config_path

Resolved path to the configuration file.

analytics property

analytics

Initialized analytics tracker; raises if not yet initialized.

ap_client property

ap_client

Initialized AppleScript client; raises if not yet initialized.

cache_service property

cache_service

Initialized cache orchestrator; raises if not yet initialized.

library_snapshot_service property

library_snapshot_service

Initialized snapshot service; raises if not yet initialized.

pending_verification_service property

pending_verification_service

Initialized verification service; raises if not yet initialized.

external_api_service property

external_api_service

Initialized API orchestrator; raises if not yet initialized.

retry_handler property

retry_handler

Initialized retry handler; raises if not yet initialized.

console_logger property

console_logger

Logger for user-facing console output.

error_logger property

error_logger

Logger for error diagnostics.

analytics_logger property

analytics_logger

Logger for analytics event recording.

db_verify_logger property

db_verify_logger

Logger for database verification operations.

initialize async

initialize()

Initialize all services with async setup requirements.

Source code in src/services/dependency_container.py

async def initialize(self) -> None:
    """Initialize all services with async setup requirements."""
    self._console_logger.info("Starting async initialization of services...")

    # Load configuration first
    if self._app_config is None:
        self._load_config()

    # Configure album type detection patterns from config
    configure_album_patterns(self.app_config)

    # Construct missing service instances
    if self._analytics is None:
        loggers = LoggerContainer(
            self._console_logger,
            self._error_logger,
            self._analytics_logger,
        )
        self._analytics = Analytics(
            self.app_config,
            loggers,
        )
    if self._cache_service is None:
        self._cache_service = CacheOrchestrator(self.app_config, self._console_logger)
    if self._library_snapshot_service is None:
        self._library_snapshot_service = LibrarySnapshotService(self.app_config, self._console_logger)
    if self._pending_verification_service is None:
        self._pending_verification_service = PendingVerificationService(self.app_config, self._console_logger, self._error_logger)
    if self._api_orchestrator is None:
        self._api_orchestrator = create_external_api_orchestrator(
            self.app_config,
            self._console_logger,
            self._error_logger,
            self._analytics,
            self._cache_service,
            self._pending_verification_service,
        )

    # Initialize retry handler from typed config
    if self._retry_handler is None:
        retry_cfg = self.app_config.applescript_retry
        retry_policy = RetryPolicy(
            max_retries=retry_cfg.max_retries,
            base_delay_seconds=retry_cfg.base_delay_seconds,
            max_delay_seconds=retry_cfg.max_delay_seconds,
            jitter_range=retry_cfg.jitter_range,
            operation_timeout_seconds=retry_cfg.operation_timeout_seconds,
        )
        self._retry_handler = DatabaseRetryHandler(
            logger=self._console_logger,
            default_policy=retry_policy,
        )
        self._console_logger.debug(
            "Retry handler initialized with policy: max_retries=%d, base_delay=%.1fs",
            retry_policy.max_retries,
            retry_policy.base_delay_seconds,
        )

    # Ensure the AppleScript client is instantiated
    if self._ap_client is None:
        self._initialize_apple_script_client(self._dry_run)

    # Initialize services in the correct order
    services: list[tuple[InitializableService | None, str]] = [
        (self._library_snapshot_service, "Library Snapshot Service"),
        (self._cache_service, "Cache Service"),
        (self._pending_verification_service, "Pending Verification Service"),
        (self._api_orchestrator, "API Orchestrator"),
        (self._ap_client, "AppleScript Client"),
    ]

    for service, name in services:
        if service is not None:
            await self._initialize_service(service, name)

    # Ensure the AppleScript client is initialized
    if self._ap_client is None:
        msg = "AppleScript client must be initialized"
        raise RuntimeError(msg)

    self._console_logger.info(" All services initialized successfully")

get_analytics

get_analytics()

Return the initialized analytics tracker; raises if not yet initialized.

Source code in src/services/dependency_container.py

def get_analytics(self) -> Analytics:
    """Return the initialized analytics tracker; raises if not yet initialized."""
    if self._analytics is None:
        msg = "Analytics not initialized"
        raise RuntimeError(msg)
    return self._analytics

get_error_logger

get_error_logger()

Return the error diagnostics logger.

Source code in src/services/dependency_container.py

def get_error_logger(self) -> logging.Logger:
    """Return the error diagnostics logger."""
    return self._error_logger

get_console_logger

get_console_logger()

Return the user-facing console logger.

Source code in src/services/dependency_container.py

def get_console_logger(self) -> logging.Logger:
    """Return the user-facing console logger."""
    return self._console_logger

close async

close()

Close all services in the correct order.

Order: API first (to flush pending tasks), then cache (to persist data). This prevents API orchestrator from writing to a closed cache during shutdown.

Source code in src/services/dependency_container.py

async def close(self) -> None:
    """Close all services in the correct order.

    Order: API first (to flush pending tasks), then cache (to persist data).
    This prevents API orchestrator from writing to a closed cache during shutdown.
    """
    self._console_logger.debug("Closing %s...", LogFormat.entity("DependencyContainer"))

    # 1. FIRST: Close API orchestrator (flushes pending tasks)
    if self._api_orchestrator is not None:
        if not hasattr(self._api_orchestrator, "close"):
            self._console_logger.error("API Orchestrator does not have a 'close' method. Possible interface change or initialization error.")
        else:
            try:
                await self._api_orchestrator.close()
                self._console_logger.debug("%s closed successfully", LogFormat.entity("ExternalApiOrchestrator"))
            except (OSError, RuntimeError, asyncio.CancelledError) as e:
                self._console_logger.warning("Failed to close API Orchestrator: %s", e)

    # 2. THEN: Save and shutdown cache
    if self._cache_service is not None:
        try:
            await self._cache_service.save_all_to_disk()
            self._console_logger.debug("Cache saved successfully")
        except (OSError, TypeError, ValueError, RuntimeError) as e:
            self._console_logger.warning("Failed to save cache: %s", e)
        finally:
            try:
                await self._cache_service.shutdown()
            except (OSError, RuntimeError, asyncio.CancelledError) as e:
                self._console_logger.warning("Failed to shutdown cache services: %s", e)

    self._console_logger.debug("%s closed.", LogFormat.entity("DependencyContainer"))

shutdown

shutdown()

Clean up non-async resources and stop services.

Source code in src/services/dependency_container.py

def shutdown(self) -> None:
    """Clean up non-async resources and stop services."""
    self._console_logger.debug("Shutting down %s...", LogFormat.entity("DependencyContainer"))

    if self._listener is not None:
        self._console_logger.debug("Stopping logging listener...")
        self._listener.stop()
        self._listener = None
    self._console_logger.debug("%s shutdown complete.", LogFormat.entity("DependencyContainer"))