Skip to content

graflo.architecture.evolution.state_core

Every name here, and in the plan, spec and vocabulary submodules, resolves to :mod:graflo.architecture.evolution.aspects with a DeprecationWarning. plan_lift keeps its signature and plans both aspects, as state-core did. Removed in graflo 2.0.

Modules:

Name Description
plan
spec
vocabulary

Attributes

__all__ = ['EdgeGrounding', 'Grounding', 'LiftError', 'LiftSpec', 'plan_lift'] module-attribute

__getattr__ = forwarder(__name__, {name: ('', name) for name in ('EdgeGrounding', 'Grounding', 'LiftError', 'LiftSpec', 'plan_lift')}) module-attribute

Classes

EdgeGrounding

Bases: Grounding

A grounding addressed at one edge triple.

Source code in graflo/architecture/evolution/aspects/spec.py 
49
50
51
52
53
54
class EdgeGrounding(Grounding):
    """A grounding addressed at one edge triple."""

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(default=None, description="Relation name.")

Attributes

relation = PydanticField(default=None, description='Relation name.') class-attribute instance-attribute
source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute
target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Grounding

Bases: ConfigBaseModel

An external-vocabulary anchor a caller asserts for one element.

Source code in graflo/architecture/evolution/aspects/spec.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class Grounding(ConfigBaseModel):
    """An external-vocabulary anchor a caller asserts for one element."""

    iri: str | None = PydanticField(
        default=None, description="IRI of the concept this element denotes."
    )
    exact_match: list[str] = PydanticField(
        default_factory=list, description="IRIs asserted equivalent to it."
    )
    synonyms: list[str] = PydanticField(
        default_factory=list, description="Alternative labels an agent may meet."
    )

    @model_validator(mode="after")
    def _validate_says_something(self) -> Grounding:
        if not self.iri and not self.exact_match and not self.synonyms:
            raise ValueError(
                "a grounding must carry at least one of iri, exact_match, synonyms"
            )
        return self

Attributes

exact_match = PydanticField(default_factory=list, description='IRIs asserted equivalent to it.') class-attribute instance-attribute
iri = PydanticField(default=None, description='IRI of the concept this element denotes.') class-attribute instance-attribute
synonyms = PydanticField(default_factory=list, description='Alternative labels an agent may meet.') class-attribute instance-attribute

LiftError

Bases: ValueError

The lift cannot be planned, and the message says which declaration is missing.

Raised rather than guessed. Every case here is one where the planner can see that something is wrong but not what the right answer is -- an ungrounded type, a name it would have to overwrite, a property it was told to move that is part of the key.

Source code in graflo/architecture/evolution/aspects/base.py 
10
11
12
13
14
15
16
17
class LiftError(ValueError):
    """The lift cannot be planned, and the message says which declaration is missing.

    Raised rather than guessed. Every case here is one where the planner can see
    that something is wrong but not what the right answer is -- an ungrounded
    type, a name it would have to overwrite, a property it was told to move that
    is part of the key.
    """

LiftSpec

Bases: ConfigBaseModel

The semantic input to :func:~graflo.architecture.evolution.aspects.plan_lift.

Source code in graflo/architecture/evolution/aspects/spec.py 
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class LiftSpec(ConfigBaseModel):
    """The semantic input to :func:`~graflo.architecture.evolution.aspects.plan_lift`."""

    grounding: dict[str, Grounding] = PydanticField(
        default_factory=dict,
        description="Per-vertex grounding: ``{type_name: Grounding}``.",
    )
    edge_grounding: list[EdgeGrounding] = PydanticField(
        default_factory=list,
        description="Per-edge grounding, addressed by triple.",
    )
    identity: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Explicit identity for types that would otherwise fall back to "
            "``identity_from_all_properties``. Required only for those."
        ),
    )
    stateful: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-type mutable properties: ``{type_name: [property, ...]}``. Each "
            "named type gains a ``<Type>State`` carrying those properties over a "
            "validity interval."
        ),
    )
    observed: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Types that gain a ``<Type>Observation`` scaffold for measurements "
            "taken of them at a time."
        ),
    )
    measured: dict[str, str] = PydanticField(
        default_factory=dict,
        description=(
            "Units for existing properties, as ``{'Type.property': ucum_token}``. "
            "UCUM has no currency, so currency uses ISO-4217 (``USD``)."
        ),
    )
    provenance: bool = PydanticField(
        default=True,
        description=(
            "Lift the ``provenance`` aspect as well: ``SourceRecord`` and ``Agent`` "
            "with the edges that make 'where did this fact come from' answerable. "
            "``false`` leaves it out, as does not naming the aspect."
        ),
    )
    provenance_for: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Existing types whose facts are read from a source record: each gains "
            "a ``wasDerivedFrom`` edge to ``SourceRecord``. The state and "
            "observation scaffolds get one regardless."
        ),
    )
    mentions_for: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Existing types a source record can name: ``SourceRecord`` gains a "
            "``mentions`` edge to each."
        ),
    )
    access_for: dict[str, FieldMerge] = PydanticField(
        default_factory=dict,
        description=(
            "Vertex types that gain ``access_zones``, each with the merge mode of "
            "its zones on upsert: ``replace`` or ``union``."
        ),
    )
    access_edges: dict[str, FieldMerge] = PydanticField(
        default_factory=dict,
        description=(
            "Edge relations that gain ``access_zones``, each with the merge mode "
            "of its zones on upsert: ``replace`` or ``union``."
        ),
    )
    retire: Literal["move", "keep"] = PydanticField(
        default="move",
        description=(
            "What happens to a property named in ``stateful``. ``move`` removes it "
            "from the entity, since a fact that changes over "
            "time does not belong on the thing it is about. ``keep`` leaves it as "
            "a denormalized current value beside the history."
        ),
    )

    @model_validator(mode="after")
    def _validate_measured_addresses(self) -> LiftSpec:
        bad = sorted(key for key in self.measured if key.count(".") != 1)
        if bad:
            raise ValueError(f"measured keys must read 'Type.property', got {bad}")
        return self

    def measured_for(self, vertex: str) -> dict[str, str]:
        """``{property: unit}`` for one type."""
        return {
            key.split(".", 1)[1]: unit
            for key, unit in self.measured.items()
            if key.split(".", 1)[0] == vertex
        }

Attributes

access_edges = PydanticField(default_factory=dict, description='Edge relations that gain ``access_zones``, each with the merge mode of its zones on upsert: ``replace`` or ``union``.') class-attribute instance-attribute
access_for = PydanticField(default_factory=dict, description='Vertex types that gain ``access_zones``, each with the merge mode of its zones on upsert: ``replace`` or ``union``.') class-attribute instance-attribute
edge_grounding = PydanticField(default_factory=list, description='Per-edge grounding, addressed by triple.') class-attribute instance-attribute
grounding = PydanticField(default_factory=dict, description='Per-vertex grounding: ``{type_name: Grounding}``.') class-attribute instance-attribute
identity = PydanticField(default_factory=dict, description='Explicit identity for types that would otherwise fall back to ``identity_from_all_properties``. Required only for those.') class-attribute instance-attribute
measured = PydanticField(default_factory=dict, description="Units for existing properties, as ``{'Type.property': ucum_token}``. UCUM has no currency, so currency uses ISO-4217 (``USD``).") class-attribute instance-attribute
mentions_for = PydanticField(default_factory=list, description='Existing types a source record can name: ``SourceRecord`` gains a ``mentions`` edge to each.') class-attribute instance-attribute
observed = PydanticField(default_factory=list, description='Types that gain a ``<Type>Observation`` scaffold for measurements taken of them at a time.') class-attribute instance-attribute
provenance = PydanticField(default=True, description="Lift the ``provenance`` aspect as well: ``SourceRecord`` and ``Agent`` with the edges that make 'where did this fact come from' answerable. ``false`` leaves it out, as does not naming the aspect.") class-attribute instance-attribute
provenance_for = PydanticField(default_factory=list, description='Existing types whose facts are read from a source record: each gains a ``wasDerivedFrom`` edge to ``SourceRecord``. The state and observation scaffolds get one regardless.') class-attribute instance-attribute
retire = PydanticField(default='move', description='What happens to a property named in ``stateful``. ``move`` removes it from the entity, since a fact that changes over time does not belong on the thing it is about. ``keep`` leaves it as a denormalized current value beside the history.') class-attribute instance-attribute
stateful = PydanticField(default_factory=dict, description='Per-type mutable properties: ``{type_name: [property, ...]}``. Each named type gains a ``<Type>State`` carrying those properties over a validity interval.') class-attribute instance-attribute

Methods:

measured_for(vertex)

{property: unit} for one type.

Source code in graflo/architecture/evolution/aspects/spec.py 
151
152
153
154
155
156
157
def measured_for(self, vertex: str) -> dict[str, str]:
    """``{property: unit}`` for one type."""
    return {
        key.split(".", 1)[1]: unit
        for key, unit in self.measured.items()
        if key.split(".", 1)[0] == vertex
    }

Functions:

plan_lift(manifest, spec=None, *, aspects=DEFAULT_ASPECTS, authored=None)

The ops that lift manifest with the named aspects.

Parameters:

Name Type Description Default
manifest GraphManifest

the manifest to lift, already finish_init()-ed.

 required
spec LiftSpec | None

the semantic declarations a planner cannot infer. Required by the state and access aspects; the provenance aspect reads only provenance_for, mentions_for and provenance from it, the access aspect only access_for and access_edges.

 None
aspects Iterable[str]

which aspects to lift, by name. The state aspect's scaffolds are attached to SourceRecord when both are named.

 DEFAULT_ASPECTS
authored dict[str, Any] | None

the manifest as written, when available; see :func:~graflo.architecture.evolution.aspects.state.plan.plan_state.

 None

Returns:

Type Description
list[ManifestOp]

Ops in application order: the additive ops of every aspect, then the
list[ManifestOp]

removals, so a reviewer sees everything that is added before anything
list[ManifestOp]

that is taken away.

Raises:

Type Description
LiftError

an unknown aspect, a missing declaration, or a name the lift would mint that is already taken.
Source code in graflo/architecture/evolution/aspects/__init__.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def plan_lift(
    manifest: GraphManifest,
    spec: LiftSpec | None = None,
    *,
    aspects: Iterable[str] = DEFAULT_ASPECTS,
    authored: dict[str, Any] | None = None,
) -> list[ManifestOp]:
    """The ops that lift *manifest* with the named *aspects*.

    Args:
        manifest: the manifest to lift, already ``finish_init()``-ed.
        spec: the semantic declarations a planner cannot infer. Required by the
            ``state`` and ``access`` aspects; the ``provenance`` aspect reads
            only ``provenance_for``, ``mentions_for`` and ``provenance`` from it,
            the ``access`` aspect only ``access_for`` and ``access_edges``.
        aspects: which aspects to lift, by name. The ``state`` aspect's
            scaffolds are attached to ``SourceRecord`` when both are named.
        authored: the manifest as written, when available; see
            :func:`~graflo.architecture.evolution.aspects.state.plan.plan_state`.

    Returns:
        Ops in application order: the additive ops of every aspect, then the
        removals, so a reviewer sees everything that is added before anything
        that is taken away.

    Raises:
        LiftError: an unknown aspect, a missing declaration, or a name the lift
            would mint that is already taken.
    """
    # Deferred: apply sits beside this package, and only a composed lift needs it.
    from ..apply import apply_evolution

    chosen = {aspect.name for aspect in resolve_aspects(aspects)}
    spec = spec or LiftSpec()
    ops: list[ManifestOp] = []
    # Each aspect is planned against the manifest as the aspects before it left
    # it, so every join sees the types and fields it would otherwise only be
    # told about.
    lifted = manifest

    def _then(planned: list[ManifestOp]) -> None:
        nonlocal lifted
        ops.extend(planned)
        if planned:
            lifted = apply_evolution(lifted, planned, bump_version=False)

    state: StatePlan | None = None
    if STATE in chosen:
        state = plan_state(manifest, spec, authored=authored)
        _then(state.ops)

    if PROVENANCE in chosen and spec.provenance:
        schema = manifest.graph_schema
        existing = (
            set(schema.core_schema.vertex_config.vertex_set)
            if schema is not None
            else set()
        )
        for field, names in (
            ("provenance_for", spec.provenance_for),
            ("mentions_for", spec.mentions_for),
        ):
            unknown = sorted(set(names) - existing)
            if unknown:
                raise LiftError(
                    f"{field} names types the manifest does not have: {unknown}"
                )
        attach = [
            *sorted(spec.provenance_for),
            *(state.scaffolded if state is not None else []),
        ]
        _then(
            plan_provenance(lifted, attach=attach, mentions=sorted(spec.mentions_for))
        )

    if ACCESS in chosen:
        _then(plan_access(lifted, vertices=spec.access_for, edges=spec.access_edges))

    if SYNC in chosen:
        _then(plan_sync(lifted))

    if state is not None:
        ops += state.retire
    return ops