type_bridge.migration.info

info

Schema information container for TypeDB schema management.

SchemaInfo

SchemaInfo()

Container for organized schema information.

Initialize SchemaInfo with empty collections.

Source code in type_bridge/migration/info.py

def __init__(self):
    """Initialize SchemaInfo with empty collections."""
    self.entities: list[type[Entity]] = []
    self.relations: list[type[Relation]] = []
    self.attribute_classes: set[type[Attribute]] = set()

get_entity_by_name

get_entity_by_name(name)

Get entity by type name.

Parameters:

Name	Type	Description	Default
name	str	Entity type name	required

Returns:

Type	Description
type[Entity] | None	Entity class or None if not found

Source code in type_bridge/migration/info.py

def get_entity_by_name(self, name: str) -> type[Entity] | None:
    """Get entity by type name.

    Args:
        name: Entity type name

    Returns:
        Entity class or None if not found
    """
    for entity in self.entities:
        if entity.get_type_name() == name:
            return entity
    return None

get_relation_by_name

get_relation_by_name(name)

Get relation by type name.

Parameters:

Name	Type	Description	Default
name	str	Relation type name	required

Returns:

Type	Description
type[Relation] | None	Relation class or None if not found

Source code in type_bridge/migration/info.py

def get_relation_by_name(self, name: str) -> type[Relation] | None:
    """Get relation by type name.

    Args:
        name: Relation type name

    Returns:
        Relation class or None if not found
    """
    for relation in self.relations:
        if relation.get_type_name() == name:
            return relation
    return None

validate

validate()

Validate schema definitions for TypeDB constraints.

Raises:

Type	Description
SchemaValidationError	If schema violates TypeDB constraints

Source code in type_bridge/migration/info.py

def validate(self) -> None:
    """Validate schema definitions for TypeDB constraints.

    Raises:
        SchemaValidationError: If schema violates TypeDB constraints
    """
    # Validate entities
    for entity_model in self.entities:
        self._validate_no_duplicate_attribute_types(entity_model, entity_model.get_type_name())

    # Validate relations
    for relation_model in self.relations:
        self._validate_no_duplicate_attribute_types(
            relation_model, relation_model.get_type_name()
        )

to_typeql

to_typeql()

Generate TypeQL schema definition from collected schema information.

Base classes (with base=True) are skipped as they don't appear in TypeDB schema.

Validates the schema before generation.

Returns:

Type	Description
str	TypeQL schema definition string

Raises:

Type	Description
SchemaValidationError	If schema validation fails

Source code in type_bridge/migration/info.py

def to_typeql(self) -> str:
    """Generate TypeQL schema definition from collected schema information.

    Base classes (with base=True) are skipped as they don't appear in TypeDB schema.

    Validates the schema before generation.

    Returns:
        TypeQL schema definition string

    Raises:
        SchemaValidationError: If schema validation fails
    """
    # Validate schema before generation
    self.validate()

    lines = []

    # Define attributes first
    lines.append("define")
    lines.append("")

    # Sort attributes by name for consistent output
    sorted_attrs = sorted(self.attribute_classes, key=lambda x: x.get_attribute_name())
    for attr_class in sorted_attrs:
        lines.append(attr_class.to_schema_definition())

    lines.append("")

    # Define entities (skip base classes)
    for entity_model in self.entities:
        schema_def = entity_model.to_schema_definition()
        if schema_def is not None:  # Skip base classes
            lines.append(schema_def)
            lines.append("")

    # Define relations (skip base classes)
    for relation_model in self.relations:
        schema_def = relation_model.to_schema_definition()
        if schema_def is not None:  # Skip base classes
            lines.append(schema_def)

            # Add role player definitions
            for role_name, role in relation_model._roles.items():
                for player_type in role.player_types:
                    lines.append(
                        f"{player_type} plays {relation_model.get_type_name()}:{role.role_name};"
                    )
            lines.append("")

    return "\n".join(lines)

compare

compare(other)

Compare this schema with another schema.

Parameters:

Name	Type	Description	Default
other	SchemaInfo	Another SchemaInfo to compare against	required

Returns:

Type	Description
SchemaDiff	SchemaDiff containing all differences between the schemas

Source code in type_bridge/migration/info.py

def compare(self, other: "SchemaInfo") -> SchemaDiff:
    """Compare this schema with another schema.

    Args:
        other: Another SchemaInfo to compare against

    Returns:
        SchemaDiff containing all differences between the schemas
    """
    diff = SchemaDiff()

    # Compare entities by type name (not Python object identity)
    self_entity_by_name = {e.get_type_name(): e for e in self.entities}
    other_entity_by_name = {e.get_type_name(): e for e in other.entities}

    self_ent_names = set(self_entity_by_name.keys())
    other_ent_names = set(other_entity_by_name.keys())

    diff.added_entities = {other_entity_by_name[n] for n in other_ent_names - self_ent_names}
    diff.removed_entities = {self_entity_by_name[n] for n in self_ent_names - other_ent_names}

    # Compare entities that exist in both (by type name)
    for type_name in self_ent_names & other_ent_names:
        self_ent = self_entity_by_name[type_name]
        other_ent = other_entity_by_name[type_name]
        entity_changes = self._compare_entity(self_ent, other_ent)
        if entity_changes:
            diff.modified_entities[other_ent] = entity_changes

    # Compare relations by type name (not Python object identity)
    self_relation_by_name = {r.get_type_name(): r for r in self.relations}
    other_relation_by_name = {r.get_type_name(): r for r in other.relations}

    self_rel_names = set(self_relation_by_name.keys())
    other_rel_names = set(other_relation_by_name.keys())

    diff.added_relations = {other_relation_by_name[n] for n in other_rel_names - self_rel_names}
    diff.removed_relations = {
        self_relation_by_name[n] for n in self_rel_names - other_rel_names
    }

    # Compare relations that exist in both (by type name)
    for type_name in self_rel_names & other_rel_names:
        self_rel = self_relation_by_name[type_name]
        other_rel = other_relation_by_name[type_name]
        relation_changes = self._compare_relation(self_rel, other_rel)
        if relation_changes:
            diff.modified_relations[other_rel] = relation_changes

    # Compare attributes
    diff.added_attributes = other.attribute_classes - self.attribute_classes
    diff.removed_attributes = self.attribute_classes - other.attribute_classes

    return diff