graflo.architecture.schema.identity_uuid

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

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

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

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 in vertex_config.assigned_vertices:
        by_loc = acc_vertex.get(vname)
        if not by_loc:
            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:
                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, arango_key_mirror=False)

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

Source code in graflo/architecture/schema/identity_uuid.py

def ensure_assigned_uuids_on_docs(
    data: list[dict[str, Any]],
    *,
    preferred_field: str,
    arango_key_mirror: bool = False,
) -> None:
    """Idempotent assigned-UUID ensure for a flat doc list (writer safety net)."""
    for doc in data:
        ensure_assigned_uuid(doc, preferred_field)
        if arango_key_mirror and "_key" not in doc:
            doc["_key"] = doc[preferred_field]

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

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

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