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)

An attribute type from the database schema.

IntrospectedOwnership dataclass

IntrospectedOwnership(owner_name, attribute_name, annotations=list())

An ownership relationship between a type and an attribute.

IntrospectedRole dataclass

IntrospectedRole(name, player_types=list())

A role in a relation.

IntrospectedRelation dataclass

IntrospectedRelation(name, roles=dict())

A relation type from the database schema.

IntrospectedEntity dataclass

IntrospectedEntity(name, supertype=None)

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]

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.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")

    # Collect unique attribute types from all models
    attr_types: set[type[Attribute]] = set()
    for model in models:
        if hasattr(model, "get_owned_attributes"):
            for attr_info in model.get_owned_attributes().values():
                attr_types.add(attr_info.typ)

    # Check each attribute type
    for attr_type in attr_types:
        attr_name = attr_type.get_attribute_name()
        if type_exists(self.db, attr_name):
            schema.attributes[attr_name] = IntrospectedAttribute(
                name=attr_name,
                value_type=getattr(attr_type, "_value_type", "string"),
            )
            logger.debug(f"Found existing attribute: {attr_name}")

    # Check each model type
    for model in models:
        type_name = model.get_type_name()

        if issubclass(model, Entity) and model is not Entity:
            if type_exists(self.db, type_name):
                schema.entities[type_name] = IntrospectedEntity(name=type_name)
                logger.debug(f"Found existing entity: {type_name}")

                # Check ownerships (pass model for fallback)
                self._introspect_ownerships_for_type(schema, type_name, model)

        elif issubclass(model, Relation) and model is not Relation:
            if type_exists(self.db, type_name):
                schema.relations[type_name] = IntrospectedRelation(name=type_name)
                logger.debug(f"Found existing relation: {type_name}")

                # Check roles and role players
                self._introspect_roles_for_relation(schema, type_name, model)

    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")

    # Query all schema information
    self._introspect_entities(schema)
    self._introspect_relations(schema)
    self._introspect_attributes(schema)
    self._introspect_ownerships(schema)
    self._introspect_role_players(schema)

    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),
    }