Skip to content

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.ComposeManifestsOp, which is binary (two manifests in, one out) and is rejected by the unary dispatcher anyway. A revision applies to exactly one manifest, so composition is not a revision op.

is_revision_op(op)

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

Source code in graflo/architecture/evolution/codec.py 
160
161
162
def is_revision_op(op: ManifestOp) -> bool:
    """Whether *op* may appear in a revision (i.e. is not binary)."""
    return type(op).__name__ != "ComposeManifestsOp"

op_from_dict(payload)

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

Source code in graflo/architecture/evolution/codec.py 
94
95
96
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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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 
 99
100
101
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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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 
127
128
129
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 
153
154
155
156
157
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)