type_bridge.migration.state

state

Migration-state records, external-store protocol, and TypeDB default backend.

MigrationRecord dataclass

MigrationRecord(app_label, name, applied_at, checksum)

Record of an applied migration.

MigrationRunRecord dataclass

MigrationRunRecord(run_id, app_label, name, checksum, direction, status, started_at, finished_at=None, error=None, executor_ip=None, executor_mac=None)

Record of one migration execution attempt.

MigrationState dataclass

MigrationState(applied=list(), version='1.0')

Complete state of applied migrations.

is_applied

is_applied(app_label, name)

Check if a migration has been applied.

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required
name	str	Migration name	required

Returns:

Type	Description
bool	True if migration has been applied

Source code in type_bridge/migration/state.py

def is_applied(self, app_label: str, name: str) -> bool:
    """Check if a migration has been applied.

    Args:
        app_label: Application label
        name: Migration name

    Returns:
        True if migration has been applied
    """
    return any(r.app_label == app_label and r.name == name for r in self.applied)

add

add(record)

Add a migration record.

Parameters:

Name	Type	Description	Default
record	MigrationRecord	Migration record to add	required

Source code in type_bridge/migration/state.py

def add(self, record: MigrationRecord) -> None:
    """Add a migration record.

    Args:
        record: Migration record to add
    """
    if not self.is_applied(record.app_label, record.name):
        self.applied.append(record)

remove

remove(app_label, name)

Remove a migration record (for rollback).

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required
name	str	Migration name	required

Source code in type_bridge/migration/state.py

def remove(self, app_label: str, name: str) -> None:
    """Remove a migration record (for rollback).

    Args:
        app_label: Application label
        name: Migration name
    """
    self.applied = [
        r for r in self.applied if not (r.app_label == app_label and r.name == name)
    ]

get_latest

get_latest(app_label)

Get the most recently applied migration for an app.

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required

Returns:

Type	Description
MigrationRecord | None	Most recent migration record, or None

Source code in type_bridge/migration/state.py

def get_latest(self, app_label: str) -> MigrationRecord | None:
    """Get the most recently applied migration for an app.

    Args:
        app_label: Application label

    Returns:
        Most recent migration record, or None
    """
    app_migrations = [r for r in self.applied if r.app_label == app_label]
    return app_migrations[-1] if app_migrations else None

get_all_for_app

get_all_for_app(app_label)

Get all applied migrations for an app.

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required

Returns:

Type	Description
list[MigrationRecord]	List of migration records in application order

Source code in type_bridge/migration/state.py

def get_all_for_app(self, app_label: str) -> list[MigrationRecord]:
    """Get all applied migrations for an app.

    Args:
        app_label: Application label

    Returns:
        List of migration records in application order
    """
    return [r for r in self.applied if r.app_label == app_label]

MigrationStateStore

Bases: Protocol

State operations required by :class:MigrationExecutor.

Implement this protocol to keep applied migration state outside the target TypeDB database. Schema bootstrap and run-log reads are intentionally not part of the contract: the default :class:MigrationStateManager provides those TypeDB-specific capabilities, while embedding orchestrators may own persistence and execution-attempt logging independently.

load_state

load_state()

Load the applied migration projection used for planning.

Source code in type_bridge/migration/state.py

def load_state(self) -> MigrationState:
    """Load the applied migration projection used for planning."""
    ...

record_applied

record_applied(app_label, name, checksum)

Persist one successfully applied migration.

Source code in type_bridge/migration/state.py

def record_applied(self, app_label: str, name: str, checksum: str) -> None:
    """Persist one successfully applied migration."""
    ...

record_unapplied

record_unapplied(app_label, name)

Remove one successfully rolled-back migration.

Source code in type_bridge/migration/state.py

def record_unapplied(self, app_label: str, name: str) -> None:
    """Remove one successfully rolled-back migration."""
    ...

record_run_started

record_run_started(app_label, name, checksum, direction)

Record the start of a Python-hosted migration execution.

Source code in type_bridge/migration/state.py

def record_run_started(
    self,
    app_label: str,
    name: str,
    checksum: str,
    direction: str,
) -> MigrationRunRecord:
    """Record the start of a Python-hosted migration execution."""
    ...

record_run_finished

record_run_finished(record, status, error=None)

Record the end of a Python-hosted migration execution.

Source code in type_bridge/migration/state.py

def record_run_finished(
    self,
    record: MigrationRunRecord,
    status: str,
    error: str | None = None,
) -> MigrationRunRecord:
    """Record the end of a Python-hosted migration execution."""
    ...

MigrationStateManager

MigrationStateManager(db)

Manages migration state in TypeDB.

Applied state is stored in TypeDB as type_bridge_migration entities, and execution attempts are stored as type_bridge_migration_run entities. The storage mechanism — schema bootstrap, reads, and writes — lives in Rust behind the MigrationStateStore seam; this class is a thin facade that delegates to a Rust-owned PyMigrationStateManager and assembles Python dataclasses for callers.

Example

manager = MigrationStateManager(db) state = manager.load_state()

if not state.is_applied("myapp", "0001_initial"): # Apply migration... manager.record_applied("myapp", "0001_initial", "abc123")

Initialize state manager.

Parameters:

Name	Type	Description	Default
db	Database	Database connection	required

Source code in type_bridge/migration/state.py

def __init__(self, db: Database):
    """Initialize state manager.

    Args:
        db: Database connection
    """
    self.db = db
    self._state: MigrationState | None = None
    self._rust_manager: Any = None

ensure_schema

ensure_schema()

Ensure the migration tracking schema exists in TypeDB.

Idempotent: the Rust state store no-ops when the schema is already ensured, so no second Python-side latch is kept here.

Source code in type_bridge/migration/state.py

def ensure_schema(self) -> None:
    """Ensure the migration tracking schema exists in TypeDB.

    Idempotent: the Rust state store no-ops when the schema is already
    ensured, so no second Python-side latch is kept here.
    """
    self._manager.ensure_schema()

load_state

load_state()

Load migration state from TypeDB.

Returns:

Type	Description
MigrationState	Current migration state

Source code in type_bridge/migration/state.py

def load_state(self) -> MigrationState:
    """Load migration state from TypeDB.

    Returns:
        Current migration state
    """
    self.ensure_schema()

    state = MigrationState()
    # Let a real backend error surface; ensure_schema() above guarantees the
    # store exists, so load_applied returns [] (not an error) on first run.
    # Swallowing here would silently mask a Rust-side failure as empty state.
    for row in self._manager.load_applied():
        applied = row.get("applied_at")
        state.add(
            MigrationRecord(
                app_label=str(row["app_label"]),
                name=str(row["name"]),
                applied_at="" if applied is None else str(applied),
                checksum=str(row["checksum"]),
            )
        )

    self._state = state
    return state

load_runs

load_runs()

Load the migration execution run log from TypeDB.

Source code in type_bridge/migration/state.py

def load_runs(self) -> list[MigrationRunRecord]:
    """Load the migration execution run log from TypeDB."""
    self.ensure_schema()

    runs: list[MigrationRunRecord] = []
    for row in self._manager.load_runs():
        runs.append(
            MigrationRunRecord(
                run_id=str(row["run_id"]),
                app_label=str(row["app_label"]),
                name=str(row["name"]),
                checksum=str(row["checksum"]),
                direction=str(row["direction"]),
                status=str(row["status"]),
                started_at=str(row["started_at"]),
                finished_at=_optional_str(row.get("finished_at")),
                error=_optional_str(row.get("error")),
                executor_ip=_optional_str(row.get("executor_ip")),
                executor_mac=_optional_str(row.get("executor_mac")),
            )
        )
    return runs

record_applied

record_applied(app_label, name, checksum)

Record that a migration was applied.

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required
name	str	Migration name	required
checksum	str	Migration content hash	required

Source code in type_bridge/migration/state.py

def record_applied(self, app_label: str, name: str, checksum: str) -> None:
    """Record that a migration was applied.

    Args:
        app_label: Application label
        name: Migration name
        checksum: Migration content hash
    """
    self.ensure_schema()

    applied_at = datetime.now(UTC)
    applied_at_str = applied_at.strftime("%Y-%m-%dT%H:%M:%S.%f")

    self._manager.record_applied(
        {
            "app_label": app_label,
            "name": name,
            "checksum": checksum,
            "applied_at": applied_at_str,
        }
    )

    logger.info(f"Recorded migration: {app_label}.{name}")

    # Update local state
    if self._state:
        self._state.add(
            MigrationRecord(
                app_label=app_label,
                name=name,
                applied_at=applied_at.isoformat(),
                checksum=checksum,
            )
        )

record_unapplied

record_unapplied(app_label, name)

Record that a migration was rolled back.

Parameters:

Name	Type	Description	Default
app_label	str	Application label	required
name	str	Migration name	required

Source code in type_bridge/migration/state.py

def record_unapplied(self, app_label: str, name: str) -> None:
    """Record that a migration was rolled back.

    Args:
        app_label: Application label
        name: Migration name
    """
    self.ensure_schema()

    self._manager.record_unapplied(app_label, name)

    logger.info(f"Removed migration record: {app_label}.{name}")

    # Update local state
    if self._state:
        self._state.remove(app_label, name)

record_run_started

record_run_started(app_label, name, checksum, direction)

Record that one migration execution attempt started.

Source code in type_bridge/migration/state.py

def record_run_started(
    self,
    app_label: str,
    name: str,
    checksum: str,
    direction: str,
) -> MigrationRunRecord:
    """Record that one migration execution attempt started."""
    if direction not in {"apply", "rollback"}:
        raise ValueError(f"Unsupported migration run direction: {direction}")

    started_at = _timestamp_now()
    executor_ip, executor_mac = _executor_info()
    record = MigrationRunRecord(
        run_id=str(uuid.uuid4()),
        app_label=app_label,
        name=name,
        checksum=checksum,
        direction=direction,
        status="started",
        started_at=started_at,
        executor_ip=executor_ip,
        executor_mac=executor_mac,
    )
    self._record_run(record)
    return record

record_run_finished

record_run_finished(record, status, error=None)

Record that a migration execution attempt finished.

Source code in type_bridge/migration/state.py

def record_run_finished(
    self,
    record: MigrationRunRecord,
    status: str,
    error: str | None = None,
) -> MigrationRunRecord:
    """Record that a migration execution attempt finished."""
    if status not in {"succeeded", "failed"}:
        raise ValueError(f"Unsupported migration run status: {status}")

    finished = MigrationRunRecord(
        run_id=record.run_id,
        app_label=record.app_label,
        name=record.name,
        checksum=record.checksum,
        direction=record.direction,
        status=status,
        started_at=record.started_at,
        finished_at=_timestamp_now(),
        error=error,
        executor_ip=record.executor_ip,
        executor_mac=record.executor_mac,
    )
    self._record_run(finished)
    return finished