type_bridge.migration

migration

Read-only archive migration recovery and introspection surfaces.

Canonical Split-YAML workspace migrations own all new planning and writes. The Python package retains only frozen-history loading, state inspection, sidecar conversion, and schema introspection needed for one-way adoption.

MIGRATION_STATE_SCHEMA module-attribute

MIGRATION_STATE_SCHEMA = _label_projection(migration_state_schema())

Immutable labels for all schema objects owned by TypeBridge migration state.

SchemaConflictError

SchemaConflictError(diff, message=None)

Bases: Exception

A retained conflict diagnostic for existing compatibility callers.

Source code in type_bridge/migration/exceptions.py

def __init__(self, diff: Any, message: str | None = None) -> None:
    self.diff = diff
    super().__init__(message or "Schema conflict detected")

has_breaking_changes

has_breaking_changes()

Report whether the supplied historical diff has breaking members.

Source code in type_bridge/migration/exceptions.py

def has_breaking_changes(self) -> bool:
    """Report whether the supplied historical diff has breaking members."""
    return any(
        bool(getattr(self.diff, name, None))
        for name in (
            "removed_entities",
            "removed_relations",
            "removed_attributes",
            "modified_attributes",
            "modified_entities",
            "modified_relations",
        )
    )

SchemaValidationError

Bases: Exception

A retained schema-validation diagnostic.

IntrospectedAttribute dataclass

IntrospectedAttribute(name, value_type, parent_type=None, is_abstract=False, is_independent=False, regex=None, allowed_values=None, range=None, doc=None, meta=dict())

An attribute type from the database schema.

IntrospectedEntity dataclass

IntrospectedEntity(name, supertype=None, is_abstract=False, doc=None, meta=dict())

An entity type from the database schema.

IntrospectedOwnership dataclass

IntrospectedOwnership(owner_name, attribute_name, annotations=list(), doc=None, meta=dict())

An ownership relationship between a type and an attribute.

IntrospectedRelation dataclass

IntrospectedRelation(name, roles=dict(), supertype=None, is_abstract=False, doc=None, meta=dict())

A relation type from the database schema.

IntrospectedRole dataclass

IntrospectedRole(name, player_types=list(), cardinality=None, doc=None, meta=dict())

A role in a relation.

IntrospectedSchema dataclass

IntrospectedSchema(entities=dict(), relations=dict(), attributes=dict(), ownerships=list())

Complete introspected schema from TypeDB database.

This is a database-centric view of the schema that can be compared against Python model definitions.

is_empty

is_empty()

Check if the schema is empty (no custom types).

Source code in type_bridge/migration/introspection.py

def is_empty(self) -> bool:
    """Check if the schema is empty (no custom types)."""
    # Filter out built-in types
    custom_entities = {k: v for k, v in self.entities.items() if k not in ("entity",)}
    custom_relations = {k: v for k, v in self.relations.items() if k not in ("relation",)}
    custom_attrs = {k: v for k, v in self.attributes.items() if k not in ("attribute",)}

    return not (custom_entities or custom_relations or custom_attrs)

get_entity_names

get_entity_names()

Get names of all custom entity types.

Source code in type_bridge/migration/introspection.py

def get_entity_names(self) -> set[str]:
    """Get names of all custom entity types."""
    return {k for k in self.entities.keys() if k != "entity"}

get_relation_names

get_relation_names()

Get names of all custom relation types.

Source code in type_bridge/migration/introspection.py

def get_relation_names(self) -> set[str]:
    """Get names of all custom relation types."""
    return {k for k in self.relations.keys() if k != "relation"}

get_attribute_names

get_attribute_names()

Get names of all custom attribute types.

Source code in type_bridge/migration/introspection.py

def get_attribute_names(self) -> set[str]:
    """Get names of all custom attribute types."""
    return {k for k in self.attributes.keys() if k != "attribute"}

get_ownerships_for

get_ownerships_for(owner_name)

Get all ownerships for a specific owner type.

Source code in type_bridge/migration/introspection.py

def get_ownerships_for(self, owner_name: str) -> list[IntrospectedOwnership]:
    """Get all ownerships for a specific owner type."""
    return [o for o in self.ownerships if o.owner_name == owner_name]

from_rust_schema_info classmethod

from_rust_schema_info(info)

Build the compatibility DTO from Rust SchemaInfo live introspection.

Source code in type_bridge/migration/introspection.py

@classmethod
def from_rust_schema_info(cls, info: dict) -> IntrospectedSchema:
    """Build the compatibility DTO from Rust ``SchemaInfo`` live introspection."""
    schema = cls()

    for attr_name, attr in info.get("attributes", {}).items():
        schema.attributes[attr_name] = IntrospectedAttribute(
            name=attr_name,
            value_type=attr.get("value_type", "string"),
            parent_type=attr.get("parent_type"),
            is_abstract=bool(attr.get("is_abstract", False)),
            is_independent=bool(attr.get("is_independent", False)),
            regex=attr.get("regex"),
            allowed_values=list(attr["allowed_values"])
            if attr.get("allowed_values") is not None
            else None,
            range=attr.get("range"),
            doc=attr.get("doc"),
            meta=dict(attr.get("meta", {}) or {}),
        )

    for entity_name, entity in info.get("entities", {}).items():
        schema.entities[entity_name] = IntrospectedEntity(
            name=entity_name,
            supertype=entity.get("parent_type"),
            is_abstract=bool(entity.get("is_abstract", False)),
            doc=entity.get("doc"),
            meta=dict(entity.get("meta", {}) or {}),
        )
        schema._add_ownerships_from_rust_entry(entity_name, entity)

    for relation_name, relation in info.get("relations", {}).items():
        schema.relations[relation_name] = IntrospectedRelation(
            name=relation_name,
            supertype=relation.get("parent_type"),
            is_abstract=bool(relation.get("is_abstract", False)),
            doc=relation.get("doc"),
            meta=dict(relation.get("meta", {}) or {}),
        )
        for role in relation.get("roles", []):
            role_name = role["role_name"]
            schema.relations[relation_name].roles[role_name] = IntrospectedRole(
                name=role_name,
                player_types=list(role.get("player_type_names", [])),
                cardinality=role.get("cardinality"),
                doc=role.get("doc"),
                meta=dict(role.get("meta", {}) or {}),
            )
        schema._add_ownerships_from_rust_entry(relation_name, relation)

    return schema

to_rust_schema_info

to_rust_schema_info()

Serialize introspected database schema to the Rust SchemaInfo dict shape.

Source code in type_bridge/migration/introspection.py

def to_rust_schema_info(self) -> dict:
    """Serialize introspected database schema to the Rust ``SchemaInfo`` dict shape."""
    info: dict = {"entities": {}, "relations": {}, "attributes": {}}

    for attr_name, attr in self.attributes.items():
        if attr_name == "attribute":
            continue
        info["attributes"][attr_name] = {
            "attr_name": attr_name,
            "value_type": _rust_value_type(attr.value_type),
        }
        _add_attribute_metadata(info["attributes"][attr_name], attr)

    for entity_name, entity in self.entities.items():
        if entity_name == "entity":
            continue
        info["entities"][entity_name] = {
            "type_name": entity_name,
            "is_abstract": entity.is_abstract,
            "parent_type": entity.supertype,
            "owned_attributes": self._owned_attributes_for(entity_name, info),
        }
        _add_doc_meta(info["entities"][entity_name], entity.doc, entity.meta)

    for relation_name, relation in self.relations.items():
        if relation_name == "relation":
            continue
        info["relations"][relation_name] = {
            "type_name": relation_name,
            "is_abstract": relation.is_abstract,
            "parent_type": relation.supertype,
            "owned_attributes": self._owned_attributes_for(relation_name, info),
            "roles": [_role_entry_to_rust(role) for role in relation.roles.values()],
        }
        _add_doc_meta(info["relations"][relation_name], relation.doc, relation.meta)

    return info

SchemaIntrospector

SchemaIntrospector(db)

Introspects TypeDB database schema.

Queries the database to discover all types, attributes, ownerships, and relations defined in the schema.

Example

introspector = SchemaIntrospector(db) schema = introspector.introspect()

print(f"Found {len(schema.entities)} entities") print(f"Found {len(schema.relations)} relations") print(f"Found {len(schema.attributes)} attributes")

Initialize introspector.

Parameters:

Name	Type	Description	Default
db	Database	Database connection	required

Source code in type_bridge/migration/introspection.py

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

    Args:
        db: Database connection
    """
    self.db = db

introspect_for_models

introspect_for_models(models)

Introspect database schema for specific model types.

This is the TypeDB 3.x compatible approach that checks each model type individually instead of enumerating all types.

Parameters:

Name	Type	Description	Default
models	list[type[_QueryEntity] | type[_QueryRelation]]	List of model classes to check	required

Returns:

Type	Description
IntrospectedSchema	IntrospectedSchema with info about existing types

Source code in type_bridge/migration/introspection.py

def introspect_for_models(
    self, models: list[type[Entity] | type[Relation]]
) -> IntrospectedSchema:
    """Introspect database schema for specific model types.

    This is the TypeDB 3.x compatible approach that checks each
    model type individually instead of enumerating all types.

    Args:
        models: List of model classes to check

    Returns:
        IntrospectedSchema with info about existing types
    """
    from type_bridge._rust_runtime import introspect_schema
    from type_bridge.models.entity import _QueryEntity as Entity
    from type_bridge.models.relation import _QueryRelation as Relation

    schema = IntrospectedSchema()

    if not self.db.database_exists():
        logger.debug("Database does not exist, returning empty schema")
        return schema

    logger.info(f"Introspecting database schema for {len(models)} model types")

    live_schema = IntrospectedSchema.from_rust_schema_info(introspect_schema(self.db))
    schema = _filter_schema_for_models(live_schema, models, Entity, Relation)

    logger.info(
        f"Introspected: {len(schema.entities)} entities, "
        f"{len(schema.relations)} relations, "
        f"{len(schema.attributes)} attributes"
    )

    return schema

introspect

introspect()

Query TypeDB schema and return structured info.

Returns:

Type	Description
IntrospectedSchema	IntrospectedSchema with all discovered types

Source code in type_bridge/migration/introspection.py

def introspect(self) -> IntrospectedSchema:
    """Query TypeDB schema and return structured info.

    Returns:
        IntrospectedSchema with all discovered types
    """
    schema = IntrospectedSchema()

    if not self.db.database_exists():
        logger.debug("Database does not exist, returning empty schema")
        return schema

    logger.info("Introspecting database schema")

    from type_bridge._rust_runtime import introspect_schema

    schema = IntrospectedSchema.from_rust_schema_info(introspect_schema(self.db))

    logger.info(
        f"Introspected: {len(schema.entities)} entities, "
        f"{len(schema.relations)} relations, "
        f"{len(schema.attributes)} attributes"
    )

    return schema

LoadedMigration dataclass

LoadedMigration(migration, path, checksum, *, execution_spec=None, source_sha256=None, execution_sidecar_sha256=None, execution_sidecar_json=None, execution_sidecar_entry=None)

A migration loaded from a file.

Attributes:

Name	Type	Description
migration	_ArchivedMigration	The Migration instance
path	Path	Path to the migration file
checksum	str	SHA256 hash of file content (first 16 chars)
execution_spec	dict[str, Any] | None	Optional pre-lowered MigrationSpec dict loaded from the JSON sidecar. Present only for generated migrations that carry a .json sibling; None for archived hand-authored files. Keyword-only so existing positional construction sites are unaffected.

MigrationLoader

MigrationLoader(migrations_dir, *, use_sidecars=True, adoption_limits=False, directory_authority=None, adoption_import_dir=None)

Loads migration files from a directory.

Migration files must follow the naming pattern: NNNN_*.py where NNNN is a 4-digit number (e.g., 0001_initial.py, 0002_add_company.py)

Example

loader = MigrationLoader(Path("migrations")) migrations = loader.discover()

for loaded in migrations: print(f"{loaded.migration.name}: {loaded.checksum}")

Initialize loader.

Parameters:

Name	Type	Description	Default
migrations_dir	Path	Directory containing migration files	required
use_sidecars	bool	Prefer checked execution sidecars when present. Adoption metadata generation preserves this released behavior: a retained valid sidecar is authoritative and prevents Python execution; only a source with no sidecar is imported.	True
adoption_limits	bool	Apply the bounded, no-follow reader used only by adoption metadata generation. The released loader path stays byte-for-byte compatible with its unbounded glob and Path.read_text() behavior.	False
directory_authority	AdoptionDirectoryAuthority | None	Retained adoption-only directory capability. When absent, an adoption-limited discovery retains one for the duration of that discovery. Ordinary V1 discovery ignores it.	None
adoption_import_dir	Path | None	Private retained mirror of migrations_dir. Required for adoption discovery so checksum decoding, module execution, and package imports all consume the same captured bytes without writing bytecode into the retained authority.	None

Source code in type_bridge/migration/loader.py

def __init__(
    self,
    migrations_dir: Path,
    *,
    use_sidecars: bool = True,
    adoption_limits: bool = False,
    directory_authority: AdoptionDirectoryAuthority | None = None,
    adoption_import_dir: Path | None = None,
):
    """Initialize loader.

    Args:
        migrations_dir: Directory containing migration files
        use_sidecars: Prefer checked execution sidecars when present.
            Adoption metadata generation preserves this released behavior:
            a retained valid sidecar is authoritative and prevents Python
            execution; only a source with no sidecar is imported.
        adoption_limits: Apply the bounded, no-follow reader used only by
            adoption metadata generation. The released loader path stays
            byte-for-byte compatible with its unbounded ``glob`` and
            ``Path.read_text()`` behavior.
        directory_authority: Retained adoption-only directory capability.
            When absent, an adoption-limited discovery retains one for the
            duration of that discovery. Ordinary V1 discovery ignores it.
        adoption_import_dir: Private retained mirror of ``migrations_dir``.
            Required for adoption discovery so checksum decoding, module
            execution, and package imports all consume the same captured
            bytes without writing bytecode into the retained authority.
    """
    self.migrations_dir = migrations_dir
    self.use_sidecars = use_sidecars
    self.adoption_limits = adoption_limits
    self._directory_authority = directory_authority
    self._adoption_import_dir = adoption_import_dir
    self._history_bytes = 0
    self._ignored_sources: list[IgnoredMigrationSource] = []
    self._adoption_entries: dict[str, AdoptionDirectoryEntry] = {}

ignored_sources property

ignored_sources

Return ignored-source evidence captured by the last discovery.

This is populated only by the adoption-limited trusted-reader path. Ordinary released discovery retains its historical return contract.

adoption_entries property

adoption_entries

Return captured source revisions from the last adoption discovery.

discover

discover()

Discover all migration files in order.

Returns:

Type	Description
list[LoadedMigration]	List of loaded migrations, sorted by filename

Source code in type_bridge/migration/loader.py

def discover(self) -> list[LoadedMigration]:
    """Discover all migration files in order.

    Returns:
        List of loaded migrations, sorted by filename
    """
    self._ignored_sources = []
    self._adoption_entries = {}
    if self.adoption_limits and self._adoption_import_dir is None:
        if not self.migrations_dir.exists():
            logger.debug(f"Migrations directory does not exist: {self.migrations_dir}")
            return []
        from type_bridge.migration._adoption_import import (
            RetainedImportError,
            retained_import_mirror,
        )

        owned = self._directory_authority is None
        authority = self._directory_authority
        if authority is None:
            try:
                authority = AdoptionDirectoryAuthority.open(self.migrations_dir)
            except OSError as error:
                raise MigrationLoadError(
                    "Failed to retain migrations directory authority"
                ) from error
        revision = authority.directory_revision()
        try:
            try:
                with retained_import_mirror(
                    authority,
                    self.migrations_dir,
                    revision,
                ) as mirror:
                    retained_loader = MigrationLoader(
                        self.migrations_dir,
                        use_sidecars=self.use_sidecars,
                        adoption_limits=True,
                        directory_authority=authority,
                        adoption_import_dir=mirror.package_dir,
                    )
                    migrations = retained_loader.discover()
                    self._ignored_sources = list(retained_loader.ignored_sources)
                    self._adoption_entries = retained_loader.adoption_entries
                    return migrations
            except RetainedImportError as error:
                raise MigrationLoadError(str(error)) from error
        finally:
            if owned:
                authority.close()

    owned_authority: AdoptionDirectoryAuthority | None = None
    if self.adoption_limits and self._directory_authority is None:
        if not self.migrations_dir.exists():
            logger.debug(f"Migrations directory does not exist: {self.migrations_dir}")
            return []
        try:
            owned_authority = AdoptionDirectoryAuthority.open(self.migrations_dir)
        except OSError as error:
            raise MigrationLoadError(
                "Failed to retain migrations directory authority"
            ) from error
        self._directory_authority = owned_authority
    elif not self.migrations_dir.exists() and not self.adoption_limits:
        logger.debug(f"Migrations directory does not exist: {self.migrations_dir}")
        return []

    try:
        authority_revision = (
            self._require_directory_authority().directory_revision()
            if self.adoption_limits
            else None
        )
        if self.adoption_limits:
            files = self._discover_adoption_sources()
        else:
            # This is the released 1.5.x discovery contract. In particular,
            # unrelated directory entries are filtered by glob rather than
            # counted toward a new resource ceiling.
            files = sorted(self.migrations_dir.glob(self.MIGRATION_PATTERN))
        migrations: list[LoadedMigration] = []
        self._history_bytes = 0

        for path in files:
            try:
                loaded = self._load_migration_file(path)
                if loaded:
                    migrations.append(loaded)
            except Exception as e:
                logger.error(f"Failed to load migration {path}: {e}")
                raise MigrationLoadError(f"Failed to load migration {path}: {e}") from e

        if authority_revision is not None:
            self._require_directory_authority().require_directory_revision(authority_revision)
        logger.debug(f"Discovered {len(migrations)} migration(s) in {self.migrations_dir}")
        return migrations
    except AdoptionDirectoryError as error:
        raise MigrationLoadError(
            "Migration authority changed during bounded adoption discovery"
        ) from error
    finally:
        if owned_authority is not None:
            owned_authority.close()
            self._directory_authority = None

get_by_name

get_by_name(name)

Get a specific migration by name.

Parameters:

Name	Type	Description	Default
name	str	Migration name (e.g., "0001_initial")	required

Returns:

Type	Description
LoadedMigration | None	LoadedMigration or None if not found

Source code in type_bridge/migration/loader.py

def get_by_name(self, name: str) -> LoadedMigration | None:
    """Get a specific migration by name.

    Args:
        name: Migration name (e.g., "0001_initial")

    Returns:
        LoadedMigration or None if not found
    """
    for loaded in self.discover():
        if loaded.migration.name == name:
            return loaded
    return None

get_by_number

get_by_number(number)

Get a specific migration by number.

Parameters:

Name	Type	Description	Default
number	int	Migration number (e.g., 1 for 0001_initial)	required

Returns:

Type	Description
LoadedMigration | None	LoadedMigration or None if not found

Source code in type_bridge/migration/loader.py

def get_by_number(self, number: int) -> LoadedMigration | None:
    """Get a specific migration by number.

    Args:
        number: Migration number (e.g., 1 for 0001_initial)

    Returns:
        LoadedMigration or None if not found
    """
    prefix = f"{number:04d}_"
    for loaded in self.discover():
        if loaded.migration.name.startswith(prefix):
            return loaded
    return None

get_next_number

get_next_number()

Get the next available migration number.

Returns:

Type	Description
int	Next migration number (1 if no migrations exist)

Source code in type_bridge/migration/loader.py

def get_next_number(self) -> int:
    """Get the next available migration number.

    Returns:
        Next migration number (1 if no migrations exist)
    """
    migrations = self.discover()
    if not migrations:
        return 1

    # Extract numbers from existing migrations
    numbers = []
    for loaded in migrations:
        try:
            num = int(loaded.migration.name[:4])
            numbers.append(num)
        except (ValueError, IndexError):
            pass

    return max(numbers) + 1 if numbers else 1

validate_dependencies

validate_dependencies()

Validate that all migration dependencies are satisfied.

Returns:

Type	Description
list[str]	List of error messages (empty if valid)

Source code in type_bridge/migration/loader.py

def validate_dependencies(self) -> list[str]:
    """Validate that all migration dependencies are satisfied.

    Returns:
        List of error messages (empty if valid)
    """
    from type_bridge.migration._lower import lower_validation_graph

    graph = lower_validation_graph(self.discover())
    errors = _rust_runtime.validate_migration_graph(graph)
    return [str(error["message"]) for error in errors]

MigrationLoadError

Bases: Exception

Error loading a migration file.

SidecarConversionError

SidecarConversionError(blockers)

Bases: Exception

A migration history cannot produce trustworthy archival artifacts.

blockers maps each migration or artifact identity to the failed trust/integrity condition. Preflight failures write nothing. A failure after the conversion journal is published can leave resumable immutable artifacts, and native adoption rejects the history until a retry clears the journal.

Source code in type_bridge/migration/sidecar.py

def __init__(self, blockers: dict[str, str]) -> None:
    self.blockers = dict(blockers)
    details = "; ".join(f"{name}: {reason}" for name, reason in sorted(self.blockers.items()))
    super().__init__(
        f"checked adoption metadata conversion did not complete. Blockers: {details}."
    )

MigrationRecord dataclass

MigrationRecord(app_label, name, applied_at, checksum)

One applied record read from the frozen ledger.

MigrationRunRecord dataclass

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

One historical migration execution record.

MigrationState dataclass

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

In-memory projection of applied archived migration records.

MigrationStateManager

MigrationStateManager(db)

Read an existing archive ledger without bootstrapping or mutating it.

Source code in type_bridge/migration/state.py

def __init__(self, db: Database):
    self.db = db
    self._rust_reader: Any = None

MigrationStateSchema dataclass

MigrationStateSchema(entities, relations, attributes, roles)

Immutable label projection of the canonical migration-state schema.

Role labels are qualified as relation:role so an application relation can use the same unqualified role name without being classified as TypeBridge infrastructure.

generate_sidecars

generate_sidecars(migrations_dir)

Generate checked JSON sidecars for every py-only migration in a directory.

Discovers the history through the released loader's frozen Python import path, lowers each executable migration with the released execution lowering, and binds every migration-shaped source to its .py checksum.

Returns:

Type	Description
list[Path]	Paths of the adoption records and executable sidecars written, in
list[Path]	source order. Empty when every required artifact already exists.

Raises:

Type	Description
SidecarConversionError	When the graph is invalid, a schema-affecting migration lacks its exact immutable snapshot, or a schema-neutral RunPython/no-op migration cannot inherit one converged parent authority. Nothing is written in that case.
MigrationLoadError	When discovery itself fails (unreadable file, stale existing sidecar, broken migration module).

Source code in type_bridge/migration/sidecar.py

def generate_sidecars(migrations_dir: Path) -> list[Path]:
    """Generate checked JSON sidecars for every py-only migration in a directory.

    Discovers the history through the released loader's frozen Python import
    path, lowers each executable migration with the released execution
    lowering, and binds every migration-shaped source to its ``.py`` checksum.

    Returns:
        Paths of the adoption records and executable sidecars written, in
        source order. Empty when every required artifact already exists.

    Raises:
        SidecarConversionError: When the graph is invalid, a schema-affecting
            migration lacks its exact immutable snapshot, or a schema-neutral
            RunPython/no-op migration cannot inherit one converged parent
            authority. Nothing is written in that case.
        MigrationLoadError: When discovery itself fails (unreadable file,
            stale existing sidecar, broken migration module).
    """
    try:
        authority = AdoptionDirectoryAuthority.open(migrations_dir)
    except OSError as error:
        raise SidecarConversionError(
            {migrations_dir.name: "migration directory authority cannot be retained"}
        ) from error
    with authority:
        try:
            return _generate_sidecars_in(migrations_dir, authority)
        except AdoptionDirectoryError as error:
            if error.errno == errno.EILSEQ:
                raise SidecarConversionError(
                    {
                        migrations_dir.name: (
                            "a migration-shaped filename is not valid UTF-8; "
                            "the native JSON identity cannot represent this released-Unix "
                            "filename safely"
                        )
                    }
                ) from error
            raise SidecarConversionError(
                {migrations_dir.name: "migration authority changed during conversion"}
            ) from error

is_migration_state_type

is_migration_state_type(*, kind, label)

Return whether label is a TypeBridge migration-state schema object.

Role labels must use the qualified relation:role form.

Source code in type_bridge/migration/state_schema.py

def is_migration_state_type(
    *,
    kind: Literal["entity", "relation", "attribute", "role"],
    label: str,
) -> bool:
    """Return whether ``label`` is a TypeBridge migration-state schema object.

    Role labels must use the qualified ``relation:role`` form.
    """
    return _rust_runtime.is_migration_state_type(kind, label)

migration_state_schema

migration_state_schema()

Return the full canonical migration-state schema descriptor from Rust.

Source code in type_bridge/migration/state_schema.py

def migration_state_schema() -> dict[str, Any]:
    """Return the full canonical migration-state schema descriptor from Rust."""
    return _rust_runtime.migration_state_schema()

without_migration_state_schema

without_migration_state_schema(schema)

Return a copy of schema without TypeBridge migration-state objects.

Source code in type_bridge/migration/state_schema.py

def without_migration_state_schema(schema: IntrospectedSchema) -> IntrospectedSchema:
    """Return a copy of ``schema`` without TypeBridge migration-state objects."""
    state_schema = MIGRATION_STATE_SCHEMA
    state_owners = state_schema.entities | state_schema.relations

    relations = {}
    for relation_name, relation in schema.relations.items():
        if relation_name in state_schema.relations:
            continue
        roles = {
            role_name: role
            for role_name, role in relation.roles.items()
            if f"{relation_name}:{role_name}" not in state_schema.roles
        }
        relations[relation_name] = replace(relation, roles=roles)

    return IntrospectedSchema(
        entities={
            name: entity
            for name, entity in schema.entities.items()
            if name not in state_schema.entities
        },
        relations=relations,
        attributes={
            name: attribute
            for name, attribute in schema.attributes.items()
            if name not in state_schema.attributes
        },
        ownerships=[
            ownership
            for ownership in schema.ownerships
            if ownership.owner_name not in state_owners
            and ownership.attribute_name not in state_schema.attributes
        ],
    )

type_exists

type_exists(db, type_name)

Check if a type exists in the database schema.

Uses a simple query to check if the type name is valid in the schema. If the type doesn't exist, the query will raise an error.

Parameters:

Name	Type	Description	Default
db	Database	Database connection	required
type_name	str	Name of the type to check (entity, relation, or attribute)	required

Returns:

Type	Description
bool	True if type exists in schema, False otherwise

Example

type_exists(db, "person") True type_exists(db, "nonexistent") False

Source code in type_bridge/migration/utils.py

def type_exists(db: "Database", type_name: str) -> bool:
    """Check if a type exists in the database schema.

    Uses a simple query to check if the type name is valid in the schema.
    If the type doesn't exist, the query will raise an error.

    Args:
        db: Database connection
        type_name: Name of the type to check (entity, relation, or attribute)

    Returns:
        True if type exists in schema, False otherwise

    Example:
        >>> type_exists(db, "person")
        True
        >>> type_exists(db, "nonexistent")
        False
    """
    query = f"""
    match $t isa {type_name};
    fetch {{ $t.* }};
    """

    try:
        with db.transaction("read") as tx:
            # If type exists, query succeeds (even with 0 results)
            # If type doesn't exist, query raises an error
            list(tx.execute(query))
            return True
    except Exception:
        return False