Skip to content

graflo.architecture.backend.fold

Upsert semantics over append-only chunks: records folded by identity.

A file backend directory is written append-only, so one entity upserted twice is stored twice. Reads fold those records into one per entity, in write order, the way a database applies the same upserts: a later record's fields replace earlier ones, except merge: union list fields, which take the union of every record's members in first-seen order, nulls dropped. The folded entity sits where its first record was.

A record whose identity is incomplete is not an upsert target and is kept as written.

Functions:

fold_edges(records, source_identity, target_identity, key_fields=(), union=())

One [source, target, properties] record per edge of records.

An edge is keyed by its endpoints' identities and the values of key_fields among its properties (the relation is the collection's own).

Source code in graflo/architecture/backend/fold.py 
 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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def fold_edges(
    records: Iterable[list[Any]],
    source_identity: Sequence[str],
    target_identity: Sequence[str],
    key_fields: Sequence[str] = (),
    union: Collection[str] = (),
) -> list[list[Any]]:
    """One ``[source, target, properties]`` record per edge of *records*.

    An edge is keyed by its endpoints' identities and the values of
    *key_fields* among its properties (the relation is the collection's own).
    """
    out: list[list[Any]] = []
    position: dict[tuple[Any, ...], int] = {}
    for record in records:
        if not isinstance(record, list) or len(record) < 2:
            out.append(record)
            continue
        source, target = record[0], record[1]
        properties = (
            record[2] if len(record) > 2 and isinstance(record[2], dict) else {}
        )
        source_key = _key(source, source_identity)
        target_key = _key(target, target_identity)
        if source_key is None or target_key is None:
            out.append(record)
            continue
        extra = tuple(str(properties.get(field)) for field in key_fields)
        key = (source_key, target_key, extra)
        at = position.get(key)
        if at is None:
            position[key] = len(out)
            out.append([dict(source), dict(target), _first(properties, union)])
        else:
            folded = out[at]
            folded[0].update(source)
            folded[1].update(target)
            merge_record(folded[2], properties, union)
    return out

fold_vertices(records, identity, union=())

One document per entity of records, keyed by the identity fields.

Source code in graflo/architecture/backend/fold.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def fold_vertices(
    records: Iterable[dict[str, Any]],
    identity: Sequence[str],
    union: Collection[str] = (),
) -> list[dict[str, Any]]:
    """One document per entity of *records*, keyed by the *identity* fields."""
    out: list[dict[str, Any]] = []
    position: dict[tuple[Any, ...], int] = {}
    for record in records:
        key = _key(record, identity)
        if key is None:
            out.append(record)
            continue
        at = position.get(key)
        if at is None:
            position[key] = len(out)
            out.append(_first(record, union))
        else:
            merge_record(out[at], record, union)
    return out

merge_record(into, record, union)

Apply record to into as an upsert: replace, and union union fields.

Source code in graflo/architecture/backend/fold.py 
22
23
24
25
26
27
28
29
30
def merge_record(
    into: dict[str, Any], record: dict[str, Any], union: Collection[str]
) -> None:
    """Apply *record* to *into* as an upsert: replace, and union *union* fields."""
    for field, value in record.items():
        if field in union:
            into[field] = union_members(into.get(field), value)
        else:
            into[field] = value