Function sqlmigrate_cmd

tortoise/cli/cli.py:632–688

(
    ctx: CLIContext,
    app_label: str | None,
    migration_name: str | None,
    backward: bool,
)

async def sqlmigrate_cmd(
    ctx: CLIContext,
    app_label: str | None,
    migration_name: str | None,
    backward: bool,
) -> None:
    config = _load_config(ctx)
    if not app_label or not migration_name:
        labels = sorted(config.apps) if config.apps else []
        available = ", ".join(labels) if labels else "(none)"
        if not app_label:
            raise utils.CLIUsageError(f"app_label is required. Available app labels: {available}")
        raise utils.CLIUsageError(
            f"migration_name is required. Usage: sqlmigrate {app_label} <migration_name>"
        )
    try:
        statements = await sqlmigrate_api(
            config=config,
            app_label=app_label,
            migration_name=migration_name,
            backward=backward,
        )
    except ValueError as exc:
        raise utils.CLIError(str(exc)) from None

    if not statements:
        print(f"{_DIM}-- (no SQL statements){_RESET}")
        return

    config_dict = config.to_dict()
    app_cfg = config_dict.get("apps", {}).get(app_label, {})
    connection_name = app_cfg.get("default_connection", "default")
    connection_url = config_dict.get("connections", {}).get(connection_name, "")
    if isinstance(connection_url, dict):
        engine = connection_url.get("engine", "")
        supports_transactional_ddl = "postgres" in engine or "psycopg" in engine
    else:
        supports_transactional_ddl = "postgres" in str(connection_url) or "psycopg" in str(
            connection_url
        )

    wrap_in_transaction = supports_transactional_ddl

    if wrap_in_transaction:
        print(f"{_DIM}BEGIN;{_RESET}")

    for statement in statements:
        if statement.startswith("--"):
            print(f"{_DIM}{statement}{_RESET}")
        else:
            if not statement.rstrip().endswith(";"):
                print(f"{statement};")
            else:
                print(statement)

    if wrap_in_transaction:
        print(f"{_DIM}COMMIT;{_RESET}")