Bases: ABC
Manages reading, ordering, and executing database migrations.
Source code in src/cloe_delta_table_manager/migrations/runner/base.py
| class MigrationRunner(ABC):
"""Manages reading, ordering, and executing database migrations."""
def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
"""Initialize the MigrationRunner.
Args:
executor: SQL executor for running statements against Databricks.
state_client: Client for reading/writing state. Defaults to LocalStateClient.
"""
self._state: MigrationState = MigrationState(migrations=[])
self._state_client: StateClient = state_client or LocalStateClient()
self._executor: SQLExecutor | None = executor
@property
def file_migrations(self) -> list[FileMigration]:
"""Get list of file-based migrations only.
Returns:
List of FileMigration instances.
"""
return [m for m in self._state.migrations if isinstance(m, FileMigration)]
@property
def pending_migrations(self) -> list[Migration]:
"""Get list of pending migrations.
Returns:
List of migrations that are in the PENDING state.
"""
return [m for m in self._state.migrations if m.status == MigrationStatus.PENDING]
@property
def applied_migrations(self) -> list[Migration]:
"""Get list of applied migrations.
Returns:
List of migrations that are in the APPLIED state.
"""
return [m for m in self._state.migrations if m.status == MigrationStatus.APPLIED]
@property
def new_migrations(self) -> list[Migration]:
"""Get list of new migrations.
Returns:
List of migrations that are in the NEW state.
"""
return [m for m in self._state.migrations if m.status == MigrationStatus.NEW]
@property
def failed_migrations(self) -> list[Migration]:
"""Get list of failed migrations.
Returns:
List of migrations that are in the FAILED state.
"""
return [m for m in self._state.migrations if m.status == MigrationStatus.FAILED]
@abstractmethod
def execute_migrations(self, *args, **kwargs) -> None:
"""Execute migrations."""
def _execute_command(self, command: str, migration: Migration) -> bool:
"""Execute a single command. Return True if any statement failed."""
assert self._executor is not None
results = self._executor.execute_statements(command)
failed_results = [r for r in results if r.failed]
for r in failed_results:
logger.error(f"Statement failed in migration {migration}: {r.error}")
return bool(failed_results)
def _execute_migration(self, migration: Migration) -> bool:
"""Execute all operations of a migration. Return True if it should be aborted."""
try:
for op in sorted(migration.operations, key=lambda op: op.index):
logger.info(f" {op}")
for command in op.commands:
if self._execute_command(command, migration):
return True
except Exception:
logger.exception(f"Unexpected error while executing migration: {migration}")
return True
return False
def _run_migrations(self, statefile_path: Path | str | None = None) -> None:
"""Run pending migrations against the database."""
graph_service = MigrationGraphService(self._state.migrations)
while (migration := graph_service.get_ready_migration()) is not None:
if self._executor is None:
raise RuntimeError("No SQL executor configured. Pass an executor when initialising the runner.")
logger.info(f"Executing migration: {migration}")
if self._execute_migration(migration):
migration.status = MigrationStatus.FAILED
logger.warning(f"Migration {migration} left as {migration.status.value}. Stopping further execution.")
break
migration.status = MigrationStatus.APPLIED
logger.info(f"Migration applied: {migration}")
if statefile_path:
self._persist_state(statefile_path)
if self.failed_migrations:
raise RuntimeError(
f"Migration phase aborted: {len(self.failed_migrations)} migration(s) failed. "
"Subsequent phases will not run."
)
def _persist_state(self, statefile_path: Path | str) -> None:
"""Append executed migrations to the existing state file."""
try:
existing = MigrationState.from_file(statefile_path, client=self._state_client)
except FileNotFoundError, ValueError:
existing = MigrationState()
existing.migrations.extend(self._state.migrations)
existing.metadata = self._state.metadata
existing.to_file(statefile_path, client=self._state_client)
|
applied_migrations
property
Get list of applied migrations.
Returns:
| Type |
Description |
list[Migration]
|
List of migrations that are in the APPLIED state.
|
failed_migrations
property
Get list of failed migrations.
Returns:
| Type |
Description |
list[Migration]
|
List of migrations that are in the FAILED state.
|
file_migrations
property
Get list of file-based migrations only.
Returns:
| Type |
Description |
list[FileMigration]
|
List of FileMigration instances.
|
new_migrations
property
Get list of new migrations.
Returns:
| Type |
Description |
list[Migration]
|
List of migrations that are in the NEW state.
|
pending_migrations
property
Get list of pending migrations.
Returns:
| Type |
Description |
list[Migration]
|
List of migrations that are in the PENDING state.
|
__init__(executor=None, state_client=None)
Initialize the MigrationRunner.
Parameters:
| Name |
Type |
Description |
Default |
executor
|
SQLExecutor | None
|
SQL executor for running statements against Databricks.
|
None
|
state_client
|
StateClient | None
|
Client for reading/writing state. Defaults to LocalStateClient.
|
None
|
Source code in src/cloe_delta_table_manager/migrations/runner/base.py
| def __init__(self, executor: SQLExecutor | None = None, state_client: StateClient | None = None):
"""Initialize the MigrationRunner.
Args:
executor: SQL executor for running statements against Databricks.
state_client: Client for reading/writing state. Defaults to LocalStateClient.
"""
self._state: MigrationState = MigrationState(migrations=[])
self._state_client: StateClient = state_client or LocalStateClient()
self._executor: SQLExecutor | None = executor
|
execute_migrations(*args, **kwargs)
abstractmethod
Execute migrations.
Source code in src/cloe_delta_table_manager/migrations/runner/base.py
| @abstractmethod
def execute_migrations(self, *args, **kwargs) -> None:
"""Execute migrations."""
|