type_bridge.migration.executor

executor

Migration executor for applying and rolling back migrations.

MigrationError

Bases: Exception

Error during migration execution.

MigrationPlan dataclass

MigrationPlan(to_apply, to_rollback)

Plan for migration execution.

Attributes:

Name	Type	Description
to_apply	list[LoadedMigration]	Migrations to apply (forward)
to_rollback	list[LoadedMigration]	Migrations to rollback (reverse)

is_empty

is_empty()

Check if plan has no operations.

Source code in type_bridge/migration/executor.py

def is_empty(self) -> bool:
    """Check if plan has no operations."""
    return not self.to_apply and not self.to_rollback

MigrationResult dataclass

MigrationResult(name, action, success, error=None, backfill=None)

Result of a migration operation.

Attributes:

Name	Type	Description
name	str	Migration name
action	str	"applied" or "rolled_back"
success	bool	Whether the operation succeeded
error	str | None	Error message if failed
backfill	list[dict[str, object]] | None	Per-step backfill counts when the migration contained a CopyAttribute op; None for pure-schema migrations (no bloat).

MigrationExecutor

MigrationExecutor(db, migrations_dir, dry_run=False, *, state_manager=None)

Executes migrations against a TypeDB database.

Handles: - Applying pending migrations - Rolling back applied migrations - Previewing migration TypeQL - Listing migration status

Example

executor = MigrationExecutor(db, Path("migrations"))

Apply all pending migrations

results = executor.migrate()

Migrate to specific version

results = executor.migrate(target="0002_add_company")

Show migration status

status = executor.showmigrations() for name, is_applied in status: print(f"[{'X' if is_applied else ' '}] {name}")

Preview TypeQL

typeql = executor.sqlmigrate("0002_add_company") print(typeql)

Initialize executor.

Parameters:

Name	Type	Description	Default
db	Database	Database connection	required
migrations_dir	Path	Directory containing migration files	required
dry_run	bool	If True, preview operations without executing	False
state_manager	MigrationStateStore | None	Optional externally owned migration-state store. When supplied, Rust migration segments use the state-free executor and never create TypeBridge's ledger schema independently in the target database. The default remains TypeDB-backed.	None

Source code in type_bridge/migration/executor.py

def __init__(
    self,
    db: Database,
    migrations_dir: Path,
    dry_run: bool = False,
    *,
    state_manager: MigrationStateStore | None = None,
):
    """Initialize executor.

    Args:
        db: Database connection
        migrations_dir: Directory containing migration files
        dry_run: If True, preview operations without executing
        state_manager: Optional externally owned migration-state store. When
            supplied, Rust migration segments use the state-free executor
            and never create TypeBridge's ledger schema independently in
            the target database. The default remains TypeDB-backed.
    """
    self.db = db
    self.migrations_dir = migrations_dir
    self.dry_run = dry_run
    self.loader = MigrationLoader(migrations_dir)
    self._uses_injected_state_manager = state_manager is not None
    self.state_manager = MigrationStateManager(db) if state_manager is None else state_manager

migrate

migrate(target=None)

Apply pending migrations.

Parameters:

Name	Type	Description	Default
target	str | None	Optional target migration name (e.g., "0002_add_company") If None, apply all pending migrations. If specified, migrate to that exact state (may rollback).	None

Returns:

Type	Description
list[MigrationResult]	List of migration results

Raises:

Type	Description
MigrationError	If migration fails

Source code in type_bridge/migration/executor.py

def migrate(self, target: str | None = None) -> list[MigrationResult]:
    """Apply pending migrations.

    Args:
        target: Optional target migration name (e.g., "0002_add_company")
               If None, apply all pending migrations.
               If specified, migrate to that exact state (may rollback).

    Returns:
        List of migration results

    Raises:
        MigrationError: If migration fails
    """
    from type_bridge.migration._lower import lower_execution_graph

    state = self.state_manager.load_state()
    all_migrations = self.loader.discover()
    self._preflight_migrations(all_migrations, state)
    plan = self._create_plan(state, all_migrations, target)

    if plan.is_empty():
        logger.info("No migrations to apply")
        return []

    if self.dry_run:
        return self._dry_run(plan)

    if _loaded_contains_run_python(all_migrations):
        self._preflight_python_plan(plan)
        return self._migrate_with_python_operations(plan, all_migrations)

    graph = lower_execution_graph(all_migrations)
    applied_records = [_applied_record_dict(record) for record in state.applied]

    runner = _rust_runtime.migration_runner_for(self.db)
    rust_results = self._apply_rust_graph(runner, graph, applied_records, target)

    checksums = {
        (loaded.migration.app_label, loaded.migration.name): loaded.checksum
        for loaded in all_migrations
    }

    results: list[MigrationResult] = []
    for rust_result in rust_results:
        result = self._record_result(rust_result, checksums)
        results.append(result)
        if not result.success:
            if result.action == "rolled_back":
                raise MigrationError(f"Rollback failed: {result.error}")
            raise MigrationError(f"Migration failed: {result.error}")

    return results

showmigrations

showmigrations()

List all migrations with their applied status.

Returns:

Type	Description
list[tuple[str, bool]]	List of (migration_name, is_applied) tuples

Source code in type_bridge/migration/executor.py

def showmigrations(self) -> list[tuple[str, bool]]:
    """List all migrations with their applied status.

    Returns:
        List of (migration_name, is_applied) tuples
    """
    state = self.state_manager.load_state()
    all_migrations = self.loader.discover()

    result: list[tuple[str, bool]] = []
    for loaded in all_migrations:
        is_applied = state.is_applied(loaded.migration.app_label, loaded.migration.name)
        result.append((loaded.migration.name, is_applied))

    return result

sqlmigrate

sqlmigrate(migration_name, reverse=False)

Preview TypeQL for a migration without executing.

Parameters:

Name	Type	Description	Default
migration_name	str	Name of the migration	required
reverse	bool	If True, show rollback TypeQL	False

Returns:

Type	Description
str	TypeQL string that would be executed

Raises:

Type	Description
MigrationError	If migration not found or not reversible

Source code in type_bridge/migration/executor.py

def sqlmigrate(self, migration_name: str, reverse: bool = False) -> str:
    """Preview TypeQL for a migration without executing.

    Args:
        migration_name: Name of the migration
        reverse: If True, show rollback TypeQL

    Returns:
        TypeQL string that would be executed

    Raises:
        MigrationError: If migration not found or not reversible
    """
    loaded = self.loader.get_by_name(migration_name)
    if loaded is None:
        raise MigrationError(f"Migration not found: {migration_name}")

    typeql = self._preview_typeql(loaded, reverse=reverse)
    if reverse and typeql is None:
        raise MigrationError(f"Migration {migration_name} is not reversible")
    return typeql or ""

plan

plan(target=None)

Get the migration plan without executing.

Parameters:

Name	Type	Description	Default
target	str | None	Optional target migration name	None

Returns:

Type	Description
MigrationPlan	MigrationPlan showing what would be applied/rolled back

Source code in type_bridge/migration/executor.py

def plan(self, target: str | None = None) -> MigrationPlan:
    """Get the migration plan without executing.

    Args:
        target: Optional target migration name

    Returns:
        MigrationPlan showing what would be applied/rolled back
    """
    state = self.state_manager.load_state()
    all_migrations = self.loader.discover()
    return self._create_plan(state, all_migrations, target)