main

CLI entry point for the Delta Table Manager.

cli(ctx, config, target, render_output)

Delta Table Manager — database schema lifecycle tool.

Source code in src/cloe_delta_table_manager/cli/main.py

@click.group()
@click.option(
    "--config",
    default="project.yaml",
    show_default=True,
    help="Path to project.yaml.",
)
@click.option(
    "--target",
    required=True,
    help="Deployment target name (must match a key under 'targets' in project.yaml).",
)
@click.option(
    "--render-output",
    default=None,
    help="Directory to write rendered template files into. Useful for debugging. "
    "When omitted a temporary directory is used and cleaned up on exit.",
)
@click.pass_context
def cli(ctx: click.Context, config: str, target: str, render_output: str | None) -> None:
    """Delta Table Manager — database schema lifecycle tool."""
    ctx.ensure_object(dict)
    ctx.obj["config_path"] = config
    ctx.obj["target"] = target
    ctx.obj["render_output"] = render_output

compare(ctx, from_state, to_state)

Show the diff between two states.

FROM_STATE and TO_STATE must each be one of: solution, database, statefile.

solution — desired schema defined in SQL files statefile — last recorded state persisted on disk database — live state read from the database (not yet supported)

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.command()
@click.argument("from_state", type=click.Choice(_STATE_KEYWORDS))
@click.argument("to_state", type=click.Choice(_STATE_KEYWORDS))
@click.pass_context
def compare(ctx: click.Context, from_state: StateKeyword, to_state: StateKeyword) -> None:
    """Show the diff between two states.

    FROM_STATE and TO_STATE must each be one of: solution, database, statefile.

    solution   — desired schema defined in SQL files
    statefile  — last recorded state persisted on disk
    database   — live state read from the database (not yet supported)
    """
    config = _resolve_config(ctx)
    target: str = ctx.obj["target"]

    try:
        current = config.build_state(from_state, target)
        desired = config.build_state(to_state, target)
    except NotImplementedError as e:
        raise click.ClickException(str(e)) from e
    except ParseError as e:
        raise click.ClickException(str(e)) from e
    except FileNotFoundError as e:
        raise click.ClickException(f"{e}\n{_init_hint(target)}") from e

    click.echo(f"Comparing {from_state} → {to_state} …\n")
    # `compare` is a preview command — render what we can and log warnings for
    # diffs the generator does not yet handle. Apply/run commands should keep
    # the default fail mode so partial migrations are never applied silently.
    migrations = MigrationGenerator.generate_migrations(
        current_state=current,
        desired_state=desired,
        on_unsupported="skip",
        platform_options=config.platform_options,
    )

    if not migrations:
        click.echo(click.style(f"No differences found — {from_state} matches {to_state}.", fg="green"))
        return

    click.echo(f"{len(migrations)} change(s):\n")
    for i, migration in enumerate(migrations):
        for op in migration.operations:
            click.echo(f"  [{i}] {op}")
    click.echo()

deploy(ctx, yes)

Deploy all pending migrations to the database.

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.command()
@click.option("--yes", "-y", is_flag=True, help="Skip the confirmation prompt.")
@click.pass_context
def deploy(ctx: click.Context, yes: bool) -> None:
    """Deploy all pending migrations to the database."""
    config = _resolve_config(ctx)
    target: str = ctx.obj["target"]

    _abort_on_validation_errors(config, target)

    click.echo(click.style("📋 Pending migrations:", bold=True))
    _print_plan(config, target)

    if not yes:
        click.confirm("\nDeploy migrations to the database?", abort=True)

    click.echo("Deploying migrations …")
    try:
        connector = config.build_databricks_connector(target)
    except ValueError as e:
        raise click.ClickException(str(e)) from e
    try:
        current_database_state = config.build_state("database", target)
        MigrationOrchestrator(
            config=config.migrations,
            target_platform=config.target_platform,
            connector=connector,
            state_client=config.build_state_client(),
            platform_options=config.platform_options,
        ).deploy(current_database_state)
    except FileNotFoundError as e:
        raise click.ClickException(f"{e}\n{_init_hint(target)}") from e
    except UnsupportedDiffError as e:
        raise click.ClickException(_unsupported_diff_message(e)) from e
    click.echo(click.style("Deployment complete.", fg="green"))

    click.echo("Refreshing database state …")
    try:
        config.refresh_database_state(target)
        click.echo(click.style("Database state refreshed.", fg="green"))
    except Exception as e:
        click.echo(
            click.style(
                f"⚠ Warning: Failed to refresh database state: {e}\n"
                f"  Run 'dtm --target {target} state refresh' manually.",
                fg="yellow",
            )
        )

generate(ctx)

Debug: write solution_state.json derived from SQL solution files.

Parses the SQL solution and serialises the resulting schema to solution_state.json. No live database connection is needed.

Use after 'state refresh' to diff the two files and spot any inconsistency between the SQL solution and the live database.

Source code in src/cloe_delta_table_manager/cli/main.py

@state.command()
@click.pass_context
def generate(ctx: click.Context) -> None:
    """Debug: write solution_state.json derived from SQL solution files.

    Parses the SQL solution and serialises the resulting schema to
    solution_state.json. No live database connection is needed.

    Use after 'state refresh' to diff the two files and spot any
    inconsistency between the SQL solution and the live database.
    """
    config = _load_config(ctx.obj["config_path"])
    target: str = ctx.obj["target"]
    _configure_logging(config.log_level)
    config = config.prepare_for_target(target, ctx.obj["render_output"])

    if not config.migrations.database_statefile_path:
        raise click.ClickException(
            f"No state_folder configured for target '{target}'. "
            "Add a 'state_folder' entry under the target in project.yaml."
        )

    click.echo(f"Generating solution state for target '{target}' …")
    try:
        solution_state = config.build_state("solution", target)
        state_client = config.build_state_client()
        state_client.write_state(config.migrations.solution_statefile_path, solution_state)
    except ParseError as e:
        raise click.ClickException(str(e)) from e
    click.echo(f"  → {config.migrations.solution_statefile_path}")
    click.echo(click.style("Done.", fg="green"))

init(ctx)

Initialise empty state files for a target (safe to re-run).

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.command()
@click.pass_context
def init(ctx: click.Context) -> None:
    """Initialise empty state files for a target (safe to re-run)."""
    config = _resolve_config(ctx)
    target: str = ctx.obj["target"]

    migrations = config.migrations
    if not migrations.database_statefile_path:
        raise click.ClickException(
            f"No state_folder configured for target '{target}'. "
            "Add a 'state_folder' entry under the target in project.yaml."
        )

    state_client = config.build_state_client()
    files: list[tuple[str, DatabaseState | MigrationState]] = [
        (migrations.database_statefile_path, DatabaseState()),
        (migrations.solution_statefile_path, DatabaseState()),
        (migrations.pre_deployment_migration_state_path, MigrationState()),
        (migrations.post_deployment_migration_state_path, MigrationState()),
        (migrations.auto_migration_state_path, MigrationState()),
    ]

    click.echo(f"Initialising state files for target '{target}' …")
    for path, empty_state in files:
        if state_client.state_exists(path):
            click.echo(f"  ~ {path} (already exists, skipped)")
        else:
            state_client.write_state(path, empty_state)
            click.echo(click.style(f"  ✓ {path}", fg="green"))
    click.echo(click.style("Done.", fg="green"))

plan(ctx, detailed)

Show pending migrations that would be run by deploy.

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.command()
@click.option(
    "--detailed",
    is_flag=True,
    help="Show migration file contents in addition to filenames.",
)
@click.pass_context
def plan(ctx: click.Context, detailed: bool) -> None:
    """Show pending migrations that would be run by deploy."""
    config = _resolve_config(ctx)
    target: str = ctx.obj["target"]

    _abort_on_validation_errors(config, target)

    _print_plan(config, target, detailed=detailed)

refresh(ctx)

Refresh the database state by crawling the live database.

Source code in src/cloe_delta_table_manager/cli/main.py

@state.command()
@click.pass_context
def refresh(ctx: click.Context) -> None:
    """Refresh the database state by crawling the live database."""
    config = _load_config(ctx.obj["config_path"])
    target: str = ctx.obj["target"]
    _configure_logging(config.log_level)
    config = config.prepare_for_target(target, ctx.obj["render_output"])

    click.echo(f"Refreshing database state for target '{target}' …")
    try:
        config.refresh_database_state(target)
    except ValueError as e:
        raise click.ClickException(str(e)) from e
    click.echo(f"  → {config.migrations.database_statefile_path}")
    click.echo(click.style("Done.", fg="green"))

state()

Manage state files.

\b refresh Crawl the live database; initialises or refreshes database_state.json. Use after out-of-band changes. generate Derive state from SQL files and write solution_state.json (debug only). No live connection needed.

Debugging tip: run 'state refresh', then 'state generate', and diff the two JSON files to spot inconsistencies between the SQL solution and the live database.

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.group()
def state() -> None:
    r"""Manage state files.

    \b
    refresh   Crawl the live database; initialises or refreshes
              database_state.json. Use after out-of-band changes.
    generate  Derive state from SQL files and write solution_state.json
              (debug only). No live connection needed.

    Debugging tip: run 'state refresh', then 'state generate', and diff
    the two JSON files to spot inconsistencies between the SQL solution
    and the live database.
    """

validate(ctx)

Validate project.yaml, SQL syntax, and the migration dependency graph.

Source code in src/cloe_delta_table_manager/cli/main.py

@cli.command()
@click.pass_context
def validate(ctx: click.Context) -> None:
    """Validate project.yaml, SQL syntax, and the migration dependency graph."""
    target: str = ctx.obj["target"]
    errors: list[str] = []

    config = _validate_project_yaml(ctx.obj["config_path"], target, errors)
    if config is None:
        click.echo(click.style(f"\n{len(errors)} validation error(s) found.", fg="red"))
        sys.exit(1)

    _validate_database_keys(config, errors)
    _validate_variable_keys(config, errors)

    if errors:
        click.echo(click.style(f"\n{len(errors)} validation error(s) found.", fg="red"))
        sys.exit(1)

    try:
        config = config.prepare_for_target(target, ctx.obj["render_output"])
    except ValueError as exc:
        msg = f"  ✗ template rendering failed: {exc}"
        click.echo(click.style(msg, fg="red"))
        click.echo(click.style("\n1 validation error(s) found.", fg="red"))
        sys.exit(1)

    _configure_logging(config.log_level)

    errors.extend(_collect_validation_errors(config))

    if errors:
        click.echo(click.style(f"\n{len(errors)} validation error(s) found.", fg="red"))
        sys.exit(1)
    click.echo(click.style("\nAll checks passed.", fg="green"))