Skip to content

serialize

serialize #

Type ↔ JSON, the type-first way, and structural equality.

A type is the source of truth: schema asks the type to describe itself, and from_schema rebuilds a type from that description — resolving scalars through the registry, or producing an opaque scalar for a name this process does not implement, so an unknown extension still round-trips.

Two types are structurally equal when their wire forms are equal. That is the only equality the storage check needs: a value may be stored at a position iff their structures match, which is what stops an Image from landing in a Thumbnail slot while letting ImageSequence and Array[Image] (equal structure, different names) interchange.

from_schema(obj) #

Rebuild a type from its wire form.

Composites recurse; a scalar resolves to its registered type, or to an opaque scalar carrying the document's kind/extensions when this process does not implement the name.

Source code in capturegraph-lib/capturegraph/types/schema/serialize.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def from_schema(obj: dict[str, Any]) -> type[CGType]:
    """Rebuild a type from its wire form.

    Composites recurse; a scalar resolves to its registered type, or to an
    opaque scalar carrying the document's ``kind``/``extensions`` when this
    process does not implement the name.
    """
    type_name = obj.get("type")
    if not isinstance(type_name, str) or not type_name:
        raise ValueError(f"Type JSON needs a non-empty 'type' string: {obj!r}")
    match type_name:
        case "array":
            return Array[from_schema(obj["element"])]
        case "struct":
            fields = {n: from_schema(f) for n, f in obj["fields"].items()}
            return type("Struct", (Struct,), {"__annotations__": fields})
        case "map":
            return Map[from_schema(obj["key"]), from_schema(obj["element"])]
        case "path":
            target = obj.get("target")
            return Path[from_schema(target)] if target is not None else Path
        case "void":
            return Void
        case _:
            known = scalar_for(type_name)
            if known is not None:
                return known
            kind = obj.get("kind")
            if kind not in ("json", "binary", None):
                raise ValueError(f"Scalar {type_name!r} has unknown kind {kind!r}")
            extensions = tuple(obj.get("extensions", ()))
            if kind == "json" and not extensions:
                extensions = ("json",)
            return opaque_scalar(type_name, kind, extensions)

schema(cgtype) #

The wire form (schema) of a type.

Source code in capturegraph-lib/capturegraph/types/schema/serialize.py 
31
32
33
def schema(cgtype: type[CGType]) -> dict[str, Any]:
    """The wire form (schema) of a type."""
    return cgtype.schema()

structural_eq(a, b) #

Whether two types describe the same on-disk structure.

Source code in capturegraph-lib/capturegraph/types/schema/serialize.py 
72
73
74
def structural_eq(a: type[CGType], b: type[CGType]) -> bool:
    """Whether two types describe the same on-disk structure."""
    return schema(a) == schema(b)