Skip to content

graflo.db.field_type_support

Backend support checks for schema field types (native or raise — no soft conversion).

UnsupportedFieldTypeError

Bases: ValueError

Raised when a field type cannot be stored natively on the target backend.

Source code in graflo/db/field_type_support.py 
17
18
19
20
21
class UnsupportedFieldTypeError(ValueError):
    """Raised when a field type cannot be stored natively on the target backend."""

    def __init__(self, message: str) -> None:
        super().__init__(message)

assert_field_type_supported(db_type, field)

Raise if field cannot be stored natively on db_type.

Soft conversions (e.g. LIST → STRING/JSON) are intentionally not performed.

Source code in graflo/db/field_type_support.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def assert_field_type_supported(db_type: DBType, field: Field) -> None:
    """Raise if ``field`` cannot be stored natively on ``db_type``.

    Soft conversions (e.g. LIST → STRING/JSON) are intentionally not performed.
    """
    if not is_list_field_type(field.type):
        return
    if db_type in _LIST_NATIVE_DBS:
        return
    label = format_field_type_label(field)
    # ``db_flavor`` reaches this function as a bare string from validated config
    # models, so the enum's ``.value`` is not always there to read.
    flavor = getattr(db_type, "value", db_type)
    raise UnsupportedFieldTypeError(
        f"Field '{field.name}' has type {label}, which cannot be stored as a "
        f"property on backend '{flavor}'. "
        "Use a backend that supports list properties, or declare an explicit "
        "STRING field if JSON encoding is intentional."
    )

assert_schema_field_types_supported(db_type, schema)

Validate every schema field against backend type support.

Source code in graflo/db/field_type_support.py 
68
69
70
71
def assert_schema_field_types_supported(db_type: DBType, schema: Schema) -> None:
    """Validate every schema field against backend type support."""
    for field in iter_schema_fields(schema):
        assert_field_type_supported(db_type, field)

iter_schema_fields(schema)

Yield all typed property fields from vertices and edges in schema.

Source code in graflo/db/field_type_support.py 
59
60
61
62
63
64
65
def iter_schema_fields(schema: Schema) -> Iterable[Field]:
    """Yield all typed property fields from vertices and edges in ``schema``."""
    for vertex in schema.core_schema.vertex_config.vertices:
        yield from vertex.properties
    for edge in schema.core_schema.edge_config.values():
        if edge.properties:
            yield from edge.properties

tigergraph_type_for_field(field)

Return a TigerGraph attribute type string (e.g. LIST<STRING>, INT).

Logical UUID is stored as STRING (TigerGraph has no native UUID type).

Source code in graflo/db/field_type_support.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def tigergraph_type_for_field(field: Field) -> str:
    """Return a TigerGraph attribute type string (e.g. ``LIST<STRING>``, ``INT``).

    Logical ``UUID`` is stored as ``STRING`` (TigerGraph has no native UUID type).
    """
    assert_field_type_supported(DBType.TIGERGRAPH, field)
    if field.type is None:
        return FieldType.STRING.value
    if is_list_field_type(field.type):
        item = field.item_type
        item_val = item.value if isinstance(item, FieldType) else str(item).upper()
        if item_val == FieldType.UUID.value:
            item_val = FieldType.STRING.value
        return f"LIST<{item_val}>"
    if isinstance(field.type, FieldType):
        if field.type == FieldType.UUID:
            return FieldType.STRING.value
        return field.type.value
    type_upper = str(field.type).upper()
    if type_upper == FieldType.UUID.value:
        return FieldType.STRING.value
    return type_upper