Skip to content

graflo.architecture.schema.identity_uuid

UUID identity helpers for assigned mode and UUID-typed natural keys.

Attributes

MINT_NAMESPACE = UUID('6f1c3a52-6a51-4f8e-9e0e-2b8c1d7a4e10') module-attribute

UUID_PATTERN = re.compile('^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE) module-attribute

Classes

Functions:

ensure_assigned_uuid(doc, field)

Fill empty field with uuid4(); validate non-empty values in place.

Never overwrites a present valid UUID.

Source code in graflo/architecture/schema/identity_uuid.py 
53
54
55
56
57
58
59
60
61
62
63
64
def ensure_assigned_uuid(doc: dict[str, Any], field: str) -> None:
    """Fill empty *field* with ``uuid4()``; validate non-empty values in place.

    Never overwrites a present valid UUID.
    """
    current = doc.get(field)
    if _identity_value_is_empty(current):
        doc[field] = str(uuid4())
        return
    doc[field] = validate_uuid_value(
        current, context=f"assigned identity field '{field}'"
    )

ensure_assigned_uuids_in_acc_vertex(acc_vertex, vertex_config)

Mint/validate assigned UUIDs on acc_vertex docs before edge assembly.

acc_vertex maps vertex name -> location -> list of VertexRep (or objects with a .vertex dict attribute).

Source code in graflo/architecture/schema/identity_uuid.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def ensure_assigned_uuids_in_acc_vertex(
    acc_vertex: Mapping[str, Any],
    vertex_config: VertexConfig,
) -> None:
    """Mint/validate assigned UUIDs on ``acc_vertex`` docs before edge assembly.

    ``acc_vertex`` maps vertex name -> location -> list of ``VertexRep`` (or
    objects with a ``.vertex`` dict attribute).
    """
    for vname, by_loc in acc_vertex.items():
        if not by_loc or not vertex_config.is_assigned(vname):
            continue
        identity_fields = vertex_config.identity_fields(vname)
        preferred = identity_fields[0] if identity_fields else "id"
        for reps in by_loc.values():
            for rep in reps:
                if isinstance(rep, VertexRep) and rep.find is not None:
                    # Found by a secondary identity: it must not get a key of
                    # its own, which would make it a new vertex.
                    continue
                doc = rep.vertex if hasattr(rep, "vertex") else rep
                if isinstance(doc, dict):
                    ensure_assigned_uuid(doc, preferred)

ensure_assigned_uuids_on_docs(data, *, preferred_field)

Idempotent assigned-UUID ensure for a flat doc list (writer safety net).

Source code in graflo/architecture/schema/identity_uuid.py 
92
93
94
95
96
97
98
99
def ensure_assigned_uuids_on_docs(
    data: list[dict[str, Any]],
    *,
    preferred_field: str,
) -> None:
    """Idempotent assigned-UUID ensure for a flat doc list (writer safety net)."""
    for doc in data:
        ensure_assigned_uuid(doc, preferred_field)

mint_uuid(schema_ref, source_id, vertex_type, ref)

The key of a keyless entity introduced as ref in source source_id.

Deterministic, so the same source cast against the same schema gives the same key; a different ref or source_id gives a new one.

Source code in graflo/architecture/schema/identity_uuid.py 
29
30
31
32
33
34
35
36
def mint_uuid(schema_ref: str, source_id: str, vertex_type: str, ref: str) -> str:
    """The key of a keyless entity introduced as *ref* in source *source_id*.

    Deterministic, so the same source cast against the same schema gives the
    same key; a different *ref* or *source_id* gives a new one.
    """
    name = f"{schema_ref}\x1f{source_id}\x1f{vertex_type}\x1f{ref}"
    return str(uuid5(MINT_NAMESPACE, name))

validate_uuid_typed_identity_fields(doc, vertex)

Validate present values on UUID-typed identity fields; do not invent.

Source code in graflo/architecture/schema/identity_uuid.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def validate_uuid_typed_identity_fields(
    doc: dict[str, Any],
    vertex: Vertex,
) -> None:
    """Validate present values on UUID-typed identity fields; do not invent."""
    prop_by_name = {f.name: f for f in vertex.properties}
    for name in vertex.identity:
        field = prop_by_name.get(name)
        if field is None or is_list_field_type(field.type):
            continue
        if field_type_value(field.type) != FieldType.UUID.value:
            continue
        current = doc.get(name)
        if _identity_value_is_empty(current):
            continue
        doc[name] = validate_uuid_value(
            current,
            context=f"vertex '{vertex.name}' identity field '{name}'",
        )

validate_uuid_value(value, *, context='UUID')

Return a normalized UUID string; raise if value is non-empty but invalid.

Source code in graflo/architecture/schema/identity_uuid.py 
43
44
45
46
47
48
49
50
def validate_uuid_value(value: Any, *, context: str = "UUID") -> str:
    """Return a normalized UUID string; raise if *value* is non-empty but invalid."""
    if _identity_value_is_empty(value):
        raise ValueError(f"{context}: expected a UUID string, got empty value")
    text = str(value)
    if not UUID_PATTERN.match(text):
        raise ValueError(f"{context}: invalid UUID value {value!r}")
    return text