graflo.architecture.evolution.codec

(De)serialization for contract operations.

The op models are already pydantic under an op-discriminated union, but the union was only ever used as a type annotation: there was no way to load a heterogeneous list of ops from YAML, so change sets could not be stored, transported or replayed. This module closes that gap.

RevisionOp is ManifestOp minus :class:~graflo.architecture.evolution.ops.MergeManifestsOp, which is binary (two manifests in, one out) and is rejected by the unary dispatcher anyway. A revision applies to exactly one manifest, so merge is not a revision op.

Attributes

RevisionOp = Annotated[RemoveVerticesOp | AddResourceTransformsOp | EnsureExtractedFieldsOp | AddResourcesOp | RemoveResourcesOp | AddVerticesOp | AddEdgesOp | RetargetEdgesOp | AddSecondaryIdentitiesOp | RemoveSecondaryIdentitiesOp | ReplaceEdgeIdentitiesOp | ChangeFieldTypesOp | AddVertexIndexesOp | RemoveVertexIndexesOp | AddEdgeIndexesOp | RemoveEdgeIndexesOp | SetBindingsOp | SetDbProfileOp | SetEdgeDirectedOp | SetVertexSemanticsOp | SetVertexDescriptionsOp | SetEdgeSemanticsOp | SetFieldSemanticsOp | MergeVerticesOp | CanonicalizeOp | RenameVertexPropertiesOp | RemoveVertexPropertiesOp | AddVertexPropertiesOp | RenameVerticesOp | RenameRelationsOp | RenameResourcesOp | RemoveEdgesOp | MergeEdgesOp | RenameEdgePropertiesOp | RemoveEdgePropertiesOp | AddEdgePropertiesOp | DeclareEdgeInversesOp | RetractEdgeInversesOp | AddInverseEdgesOp | SetNativeInversesOp | SetInverseEmissionOp | ProjectManifestOp | ReplaceIdentityOp | SanitizeOp, PydanticField(discriminator='op')] module-attribute

__all__ = ['RevisionOp', 'is_revision_op', 'op_from_dict', 'op_to_dict', 'ops_from_dicts', 'ops_from_yaml', 'ops_to_dicts', 'ops_to_yaml_str'] module-attribute

Classes

Functions:

is_revision_op(op)

Whether op may appear in a revision (i.e. is not binary).

Source code in graflo/architecture/evolution/codec.py

def is_revision_op(op: ManifestOp) -> bool:
    """Whether *op* may appear in a revision (i.e. is not binary)."""
    return not isinstance(op, MergeManifestsOp)

op_from_dict(payload)

Validate one op payload into its concrete model, keyed by op.

Source code in graflo/architecture/evolution/codec.py

def op_from_dict(payload: dict[str, Any]) -> Any:
    """Validate one op payload into its concrete model, keyed by ``op``."""
    return _op_adapter.validate_python(payload)

op_to_dict(op)

Serialize one op to a payload that is guaranteed to load back.

Compact form first (defaults dropped), then verified by re-validating it. That check is not paranoia: a nested discriminated union whose tag has a default — IdentityTarget.mode is one — loses its tag under exclude_defaults and becomes unloadable. Rather than special-casing every such field, fall back to the full form whenever the compact one does not reproduce the op.

A change set exists to be replayed exactly; silently emitting something that cannot be read back is the one failure this layer must not have.

Source code in graflo/architecture/evolution/codec.py

def op_to_dict(op: Any) -> dict[str, Any]:
    """Serialize one op to a payload that is guaranteed to load back.

    Compact form first (defaults dropped), then **verified** by re-validating
    it. That check is not paranoia: a nested discriminated union whose tag has
    a default — ``IdentityTarget.mode`` is one — loses its tag under
    ``exclude_defaults`` and becomes unloadable. Rather than special-casing
    every such field, fall back to the full form whenever the compact one does
    not reproduce the op.

    A change set exists to be replayed exactly; silently emitting something
    that cannot be read back is the one failure this layer must not have.
    """
    compact = op.to_dict(skip_defaults=True) | {"op": op.op}
    try:
        if _op_adapter.validate_python(compact) == op:
            return compact
    except Exception:
        # Any validation failure means the compact form is not loadable.
        pass
    return op.to_dict(skip_defaults=False) | {"op": op.op}

ops_from_dicts(payloads)

Validate a heterogeneous list of op payloads.

Source code in graflo/architecture/evolution/codec.py

def ops_from_dicts(payloads: list[dict[str, Any]]) -> list[Any]:
    """Validate a heterogeneous list of op payloads."""
    return _op_list_adapter.validate_python(payloads)

ops_from_yaml(source)

Load ops from a YAML file path, or from a YAML string.

Accepts either a bare list of op mappings or a mapping with an ops key.

Source code in graflo/architecture/evolution/codec.py

def ops_from_yaml(source: str | Any) -> list[Any]:
    """Load ops from a YAML file path, or from a YAML string.

    Accepts either a bare list of op mappings or a mapping with an ``ops`` key.
    """
    if isinstance(source, str) and ("\n" in source or source.lstrip().startswith("-")):
        import yaml

        payload = yaml.safe_load(source)
    else:
        payload = FileHandle.load(source)
    if isinstance(payload, dict):
        payload = payload.get("ops", [])
    if not isinstance(payload, list):
        raise ValueError(
            "expected a list of operations, or a mapping with an 'ops' key; "
            f"got {type(payload).__name__}"
        )
    return ops_from_dicts(payload)

ops_to_dicts(ops)

Serialize ops to plain dicts that load back identically.

Source code in graflo/architecture/evolution/codec.py

def ops_to_dicts(ops: list[Any]) -> list[dict[str, Any]]:
    """Serialize ops to plain dicts that load back identically."""
    return [op_to_dict(op) for op in ops]

ops_to_yaml_str(ops)

Serialize ops to a YAML list.

Source code in graflo/architecture/evolution/codec.py

def ops_to_yaml_str(ops: list[Any]) -> str:
    """Serialize ops to a YAML list."""
    import yaml

    return yaml.safe_dump(ops_to_dicts(ops), sort_keys=False, default_flow_style=False)