type_bridge.migration.introspection

introspection

TypeDB schema introspection for migration auto-generation.

This module provides functionality to introspect a TypeDB database schema and convert it to a format comparable with Python model definitions.

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.

IntrospectedOwnership dataclass

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

An ownership relationship between a type and an attribute.

IntrospectedRole dataclass

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

A role in a relation.

IntrospectedRelation dataclass

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

A relation 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.

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[Entity] | type[Relation]]	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 import Entity, 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

compare_schemas

compare_schemas(db_schema, model_names)

Compare database schema against model type names.

Parameters:

Name	Type	Description	Default
db_schema	IntrospectedSchema	Introspected database schema	required
model_names	dict[str, str]	Dict mapping model class names to TypeDB type names	required

Returns:

Type	Description
dict[str, list[str]]	Dict with 'added', 'removed', 'unchanged' lists

Source code in type_bridge/migration/introspection.py

def compare_schemas(
    db_schema: IntrospectedSchema, model_names: dict[str, str]
) -> dict[str, list[str]]:
    """Compare database schema against model type names.

    Args:
        db_schema: Introspected database schema
        model_names: Dict mapping model class names to TypeDB type names

    Returns:
        Dict with 'added', 'removed', 'unchanged' lists
    """
    db_types = (
        db_schema.get_entity_names()
        | db_schema.get_relation_names()
        | db_schema.get_attribute_names()
    )

    model_types = set(model_names.values())

    return {
        "added": list(model_types - db_types),
        "removed": list(db_types - model_types),
        "unchanged": list(db_types & model_types),
    }