graflo.architecture.evolution.state_core.plan

Attributes

__all__ = ['LiftError', 'plan_lift'] module-attribute

__getattr__ = forwarder(__name__, {'LiftError': ('', 'LiftError'), 'plan_lift': ('', 'plan_lift')}) module-attribute

Classes

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

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

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

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