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)

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

MigrationExecutor

MigrationExecutor(db, migrations_dir, dry_run=False)

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

Source code in type_bridge/migration/executor.py

def __init__(
    self,
    db: Database,
    migrations_dir: Path,
    dry_run: bool = False,
):
    """Initialize executor.

    Args:
        db: Database connection
        migrations_dir: Directory containing migration files
        dry_run: If True, preview operations without executing
    """
    self.db = db
    self.migrations_dir = migrations_dir
    self.dry_run = dry_run
    self.loader = MigrationLoader(migrations_dir)
    self.state_manager = MigrationStateManager(db)

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
    """
    state = self.state_manager.load_state()
    all_migrations = self.loader.discover()
    plan = self._create_plan(state, all_migrations, target)

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

    results: list[MigrationResult] = []

    # Rollback if needed (migrating backwards)
    for loaded in plan.to_rollback:
        result = self._rollback_one(loaded)
        results.append(result)
        if not result.success:
            raise MigrationError(f"Rollback failed: {result.error}")

    # Apply forward migrations
    for loaded in plan.to_apply:
        result = self._apply_one(loaded)
        results.append(result)
        if not result.success:
            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}")

    if reverse:
        typeql = self._generate_rollback_typeql(loaded.migration)
        if typeql is None:
            raise MigrationError(f"Migration {migration_name} is not reversible")
        return typeql
    else:
        return self._generate_apply_typeql(loaded.migration)

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)