Skip to content

graflo.architecture.schema.core

Logical graph model (vertices and edges).

Classes

CoreSchema

Bases: ConfigBaseModel

Logical graph model (A): vertices and edges.

Source code in graflo/architecture/schema/core.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class CoreSchema(ConfigBaseModel):
    """Logical graph model (A): vertices and edges."""

    vertex_config: VertexConfig = PydanticField(
        ...,
        description="Configuration for vertex collections (vertices, identities, properties).",
    )
    edge_config: EdgeConfig = PydanticField(
        ...,
        description="Configuration for edge collections (edges, weights).",
    )

    @model_validator(mode="after")
    def _init_graph(self) -> CoreSchema:
        self.finish_init()
        return self

    def finish_init(self) -> None:
        self.vertex_config.finish_init()
        self._validate_edge_vertices_defined()
        self.edge_config.finish_init(self.vertex_config)
        check_hierarchy(self.vertex_config, self.edge_config)

    def _validate_edge_vertices_defined(self) -> None:
        """Ensure all edge endpoints reference defined vertex names."""
        declared_vertices = self.vertex_config.vertex_set
        edge_vertices = self.edge_config.vertices
        undefined_vertices = edge_vertices - declared_vertices
        if undefined_vertices:
            undefined_vertices_list = sorted(undefined_vertices)
            declared_vertices_list = sorted(declared_vertices)
            raise ValueError(
                "edge_config references undefined vertices: "
                f"{undefined_vertices_list}. "
                f"Declared vertices: {declared_vertices_list}"
            )

    def remove_disconnected_vertices(self) -> set[str]:
        """Remove disconnected vertices and return removed names.

        A superclass of a connected vertex stays: it is part of that vertex's
        meaning, and removing it would leave its subclass naming nothing.
        """
        hierarchy = Hierarchy.of(self)
        connected = set(self.edge_config.vertices)
        connected |= {p for v in list(connected) for p in hierarchy.superclasses(v)}
        disconnected = self.vertex_config.vertex_set - connected
        if disconnected:
            self.vertex_config.remove_vertices(disconnected)
        return disconnected

Attributes

edge_config = PydanticField(..., description='Configuration for edge collections (edges, weights).') class-attribute instance-attribute
vertex_config = PydanticField(..., description='Configuration for vertex collections (vertices, identities, properties).') class-attribute instance-attribute

Methods:

finish_init()
Source code in graflo/architecture/schema/core.py 
31
32
33
34
35
def finish_init(self) -> None:
    self.vertex_config.finish_init()
    self._validate_edge_vertices_defined()
    self.edge_config.finish_init(self.vertex_config)
    check_hierarchy(self.vertex_config, self.edge_config)
remove_disconnected_vertices()

Remove disconnected vertices and return removed names.

A superclass of a connected vertex stays: it is part of that vertex's meaning, and removing it would leave its subclass naming nothing.

Source code in graflo/architecture/schema/core.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def remove_disconnected_vertices(self) -> set[str]:
    """Remove disconnected vertices and return removed names.

    A superclass of a connected vertex stays: it is part of that vertex's
    meaning, and removing it would leave its subclass naming nothing.
    """
    hierarchy = Hierarchy.of(self)
    connected = set(self.edge_config.vertices)
    connected |= {p for v in list(connected) for p in hierarchy.superclasses(v)}
    disconnected = self.vertex_config.vertex_set - connected
    if disconnected:
        self.vertex_config.remove_vertices(disconnected)
    return disconnected

Functions: