Skip to content

symfonic.capabilities.memory.pg_staged_snapshots

pg_staged_snapshots

Immutable private PG stage values; no public record or publication receipt.

PgStagedSnapshots dataclass 

PgStagedSnapshots(graph: tuple[PgGraphSnapshot, ...], vectors: tuple[PgVectorSnapshot, ...])

document 

document(scope, dimensions)

Validate deeply on use, including low-level dataclass tampering.

Source code in src/symfonic/capabilities/memory/pg_staged_snapshots.py 
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def document(self, scope, dimensions):
    """Validate deeply on use, including low-level dataclass tampering."""
    if (type(self) is not PgStagedSnapshots or type(scope) is not MemoryScope
            or scope.identity_version != 1 or type(dimensions) is not int
            or not 1 <= dimensions <= 4096 or type(self.graph) is not tuple
            or type(self.vectors) is not tuple or not 1 <= len(self.graph) <= 64
            or len(self.vectors) != len(self.graph)):
        refuse()
    scope.validate()
    graph, vectors = {}, {}
    for node in self.graph:
        if (type(node) is not PgGraphSnapshot or type(node.scope) is not MemoryScope
                or node.scope != scope or type(node.layer) is not MemoryLayer
                or type(node.text) is not str or "\x00" in node.text
                or type(node.properties) is not tuple
                or any(type(pair) is not tuple or len(pair) != 2
                       or any(type(item) is not str or "\x00" in item for item in pair)
                       for pair in node.properties)):
            refuse()
        key = identifier(node.record_id)
        if key in graph or len(dict(node.properties)) != len(node.properties):
            refuse()
        graph[key] = {"layer": node.layer.value, "text": node.text,
                      "properties": dict(node.properties)}
    for vector in self.vectors:
        if (type(vector) is not PgVectorSnapshot or type(vector.scope) is not MemoryScope
                or vector.scope != scope or type(vector.embedding) is not tuple
                or len(vector.embedding) != dimensions
                or any(type(item) is not float or not math.isfinite(item)
                       for item in vector.embedding)):
            refuse()
        key = identifier(vector.record_id)
        if key in vectors:
            refuse()
        vectors[key] = vector.embedding
    if graph.keys() != vectors.keys():
        refuse()
    bodies = tuple((key, {"schema": "symfonic.private.pg-stage.v1",
                         "scope": [1, scope.levels], "graph": graph[key],
                         "vector": vectors[key]}) for key in sorted(graph))
    encoded = json.dumps(bodies, ensure_ascii=False, sort_keys=True,
                         separators=(",", ":"), allow_nan=False).encode()
    if len(encoded) > 65536:
        refuse()
    return bodies