Source code for mafw.db.migration.cli_support

#  Copyright 2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""CLI support for migration service connections.

:author: Bulgheroni Antonio
"""

import contextlib
import pathlib
from collections.abc import Iterator
from typing import Any

import click

from mafw.plugin_manager import get_plugin_manager
from mafw.steering.builder import SteeringBuilder




def _echo_outcome(outcome: Any) -> None:
    """Print the human-readable messages carried by a migration outcome.

    :param outcome: The :class:`~mafw.db.migration.outcome.MigrationOutcome`.
    :type outcome: Any
    """
    for message in outcome.messages:
        click.echo(message)





@contextlib.contextmanager
def _open_migration_manager(
    steering_file: str,
    *,
    directory_override: str | None,
) -> Iterator[Any]:
    """Open a migration manager from a steering file.

    Loads the plugins that register models, reads the steering file, connects to
    the configured database, resolves the migration directory against the
    steering file's own directory (the stable base), and yields a bound
    :class:`~mafw.db.migration.service.MigrationManager`. The database connection
    is always closed on exit.

    :param steering_file: Path to the TOML steering file.
    :type steering_file: str
    :param directory_override: Optional CLI migration-directory override.
    :type directory_override: str | None
    :return: A context manager yielding the migration manager.
    :rtype: Iterator[Any]
    :raises click.ClickException: when the steering file or database is invalid.
    """
    import peewee
    from playhouse.db_url import connect

    from mafw.db.db_connection import build_connection_parameters
    from mafw.db.db_model import database_proxy
    from mafw.db.migration.directory import resolve_directory, validate_directory
    from mafw.db.migration.errors import MigrationError
    from mafw.db.migration.service import MigrationContext, MigrationManager
    from mafw.steering.environment import resolve_steering_environment

    steering_path = pathlib.Path(steering_file)
    if not steering_path.is_file():
        raise click.ClickException(f'Steering file not found: {steering_file}')

    # Models register themselves on import.
    # Only the database model modules need to be imported: discovery collects the
    # registered MAFwBaseModel subclasses, which register on import. The processor
    # and UI plugin families are irrelevant to diffing/generation/application.
    try:
        get_plugin_manager().load_plugins(('db_modules',))
    except Exception as exc:  # pragma: no cover - plugin loading is environment specific
        raise click.ClickException(f'Could not load plugins needed for migration: {exc}') from exc

    try:
        builder = SteeringBuilder.from_toml(steering_path)
    except Exception as exc:
        raise click.ClickException(f'Could not read steering file: {exc}') from exc

    db_config = builder.db_config
    environment = resolve_steering_environment(builder.to_config_dict(), steering_path).values
    conf: dict[str, Any] = dict(db_config.attributes)
    if db_config.url and 'URL' not in conf:
        conf['URL'] = db_config.url
    if db_config.authentication:
        conf['authentication'] = dict(db_config.authentication)
    if db_config.parameters:
        conf['parameters'] = {key: dict(value) for key, value in db_config.parameters.items()}
    elif db_config.pragmas:
        conf['pragmas'] = dict(db_config.pragmas)

    try:
        db_url, connection_parameters = build_connection_parameters(conf, environment)
    except Exception as exc:
        raise click.ClickException(f'Invalid database configuration: {exc}') from exc

    steering_dir_value = conf.get('migration_directory')
    if steering_dir_value is not None:
        # Expand environment variables in the configured directory, matching the
        # URL behaviour, so a steering file may use ``${VAR}`` in the path too.
        from mafw.tools.toml_tools import resolve_string

        steering_dir_value = resolve_string(str(steering_dir_value), environment)
    directory = resolve_directory(
        cli_override=directory_override,
        active_parameter=None,
        steering_value=str(steering_dir_value) if steering_dir_value is not None else None,
        base=steering_path.resolve().parent,
    )

    try:
        database = connect(db_url, **connection_parameters)  # type: ignore[no-untyped-call]
        database.connect()
    except peewee.PeeweeException as exc:
        raise click.ClickException(f'The target database is unavailable: {exc}') from exc

    database_proxy.initialize(database)
    try:
        try:
            validate_directory(directory)
        except MigrationError:
            # Defer directory validation to the operation so the outcome can
            # report it through the standard FAILURE path.
            pass
        yield MigrationManager(MigrationContext(database=database, migration_directory=directory))
    finally:
        database.close()