Skip to content

graflo.architecture.evolution.ops.field_types

Ops that change the declared type of vertex and edge fields.

Classes

ChangeFieldTypesOp

Bases: ConfigBaseModel

Set the logical type of existing vertex or edge properties.

Makes the differ's CHANGE_VERTEX_FIELD_TYPE / CHANGE_EDGE_FIELD_TYPE authorable. Targets are validated against the profile's db_flavor so an unsupported type fails here rather than at define time.

Source code in graflo/architecture/evolution/ops/field_types.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class ChangeFieldTypesOp(ConfigBaseModel):
    """Set the logical type of existing vertex or edge properties.

    Makes the differ's ``CHANGE_VERTEX_FIELD_TYPE`` / ``CHANGE_EDGE_FIELD_TYPE``
    authorable. Targets are validated against the profile's ``db_flavor`` so an
    unsupported type fails here rather than at define time.
    """

    op: Literal["change_field_types"] = "change_field_types"
    vertices: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{vertex_name: {field_name: {type, item_type}}}``.",
    )
    edges: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{relation_name: {field_name: {type, item_type}}}``.",
    )
    per_edge: list[EdgeFieldTypes] = PydanticField(
        default_factory=list,
        description="Retypes on single edge triples, including an edge with no relation.",
    )

    @model_validator(mode="after")
    def _require_a_target(self) -> ChangeFieldTypesOp:
        if not self.vertices and not self.edges and not self.per_edge:
            raise ValueError(
                "change_field_types requires at least one of vertices, edges or per_edge"
            )
        if self.per_edge:
            validate_per_edge(
                self.edges,
                self.per_edge,
                lambda entry: entry.fields,
                kind="change_field_types",
            )
        return self

    def edge_changes_for(
        self, edge_id: tuple[str, str, str | None]
    ) -> dict[str, FieldTypeSpec]:
        """The retypes that reach the edge *edge_id*, from either form."""
        out = dict(self.edges.get(edge_id[2], {})) if edge_id[2] is not None else {}
        for entry in self.per_edge:
            if entry.edge_id() == edge_id:
                out |= entry.fields
        return out

Attributes

edges = PydanticField(default_factory=dict, description='``{relation_name: {field_name: {type, item_type}}}``.') class-attribute instance-attribute
op = 'change_field_types' class-attribute instance-attribute
per_edge = PydanticField(default_factory=list, description='Retypes on single edge triples, including an edge with no relation.') class-attribute instance-attribute
vertices = PydanticField(default_factory=dict, description='``{vertex_name: {field_name: {type, item_type}}}``.') class-attribute instance-attribute

Methods:

edge_changes_for(edge_id)

The retypes that reach the edge edge_id, from either form.

Source code in graflo/architecture/evolution/ops/field_types.py 
84
85
86
87
88
89
90
91
92
def edge_changes_for(
    self, edge_id: tuple[str, str, str | None]
) -> dict[str, FieldTypeSpec]:
    """The retypes that reach the edge *edge_id*, from either form."""
    out = dict(self.edges.get(edge_id[2], {})) if edge_id[2] is not None else {}
    for entry in self.per_edge:
        if entry.edge_id() == edge_id:
            out |= entry.fields
    return out

EdgeFieldTypes

Bases: EdgeSelector

Retypes on one edge triple.

Source code in graflo/architecture/evolution/ops/field_types.py 
39
40
41
42
43
44
class EdgeFieldTypes(EdgeSelector):
    """Retypes on one edge triple."""

    fields: dict[str, FieldTypeSpec] = PydanticField(
        ..., description="``{field_name: {type, item_type}}``.", min_length=1
    )

Attributes

fields = PydanticField(..., description='``{field_name: {type, item_type}}``.', min_length=1) class-attribute instance-attribute

FieldTypeSpec

Bases: ConfigBaseModel

Target logical type for one property.

Source code in graflo/architecture/evolution/ops/field_types.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class FieldTypeSpec(ConfigBaseModel):
    """Target logical type for one property."""

    type: FieldType | None = PydanticField(
        ...,
        description="New logical field type; ``None`` clears the declared type.",
    )
    item_type: FieldType | None = PydanticField(
        default=None,
        description="Element type, required when ``type`` is ``LIST``.",
    )

    @model_validator(mode="after")
    def _validate_item_type(self) -> FieldTypeSpec:
        if self.type == FieldType.LIST and self.item_type is None:
            raise ValueError("a LIST field type requires item_type")
        if self.type != FieldType.LIST and self.item_type is not None:
            raise ValueError("item_type is only meaningful for a LIST field type")
        return self

Attributes

item_type = PydanticField(default=None, description='Element type, required when ``type`` is ``LIST``.') class-attribute instance-attribute
type = PydanticField(..., description='New logical field type; ``None`` clears the declared type.') class-attribute instance-attribute

MergeFieldTypes

Bases: ConfigBaseModel

The type of a merged property, declared on the merge by its merged names.

Keyed by the class or relation name and the property name the merged manifest carries. The merge retypes every member, on either side, that carries the property (under whatever spelling it is renamed from) before folding it, so members that disagree on a type merge into the declared one. Members that do not carry the property are left alone.

Source code in graflo/architecture/evolution/ops/field_types.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class MergeFieldTypes(ConfigBaseModel):
    """The type of a merged property, declared on the merge by its merged names.

    Keyed by the class or relation name and the property name the **merged**
    manifest carries. The merge retypes every member, on either side, that
    carries the property (under whatever spelling it is renamed from) before
    folding it, so members that disagree on a type merge into the declared one.
    Members that do not carry the property are left alone.
    """

    vertices: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{merged_vertex: {merged_property: {type, item_type}}}``.",
    )
    edges: dict[str, dict[str, FieldTypeSpec]] = PydanticField(
        default_factory=dict,
        description="``{merged_relation: {property: {type, item_type}}}``.",
    )

    @model_validator(mode="after")
    def _require_a_target(self) -> MergeFieldTypes:
        if not self.vertices and not self.edges:
            raise ValueError("field_types requires at least one of vertices or edges")
        return self

Attributes

edges = PydanticField(default_factory=dict, description='``{merged_relation: {property: {type, item_type}}}``.') class-attribute instance-attribute
vertices = PydanticField(default_factory=dict, description='``{merged_vertex: {merged_property: {type, item_type}}}``.') class-attribute instance-attribute

Functions: