graflo.db.nebula.util

NebulaGraph utility functions.

Type mapping, filter rendering, value escaping, and schema-propagation helpers.

Attributes

DEFAULT_NEBULA_TYPE = 'string' module-attribute

FIELD_TYPE_TO_NEBULA = {FieldType.INT: 'int64', FieldType.UINT: 'int64', FieldType.FLOAT: 'float', FieldType.DOUBLE: 'double', FieldType.BOOL: 'bool', FieldType.STRING: 'string', FieldType.DATETIME: 'string', FieldType.UUID: 'string'} module-attribute

NEBULA_TYPE_TO_FIELD_TYPE = {'int': FieldType.INT, 'int8': FieldType.INT, 'int16': FieldType.INT, 'int32': FieldType.INT, 'int64': FieldType.INT, 'float': FieldType.FLOAT, 'double': FieldType.DOUBLE, 'bool': FieldType.BOOL, 'string': FieldType.STRING, 'date': FieldType.DATETIME, 'time': FieldType.DATETIME, 'datetime': FieldType.DATETIME, 'timestamp': FieldType.DATETIME} module-attribute

VID_SEPARATOR = '::' module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

edge_rank(props, identity_fields)

The rank of an edge whose identity holds identity_fields besides its endpoints.

NebulaGraph tells edges of one type between the same vertices apart by rank alone, so such an edge takes a signed 64-bit digest of those values, rendered as :func:vertex_key renders identities. None (rank 0) when there are none.

Source code in graflo/db/nebula/util.py

def edge_rank(props: dict[str, Any], identity_fields: Sequence[str]) -> int | None:
    """The rank of an edge whose identity holds *identity_fields* besides its endpoints.

    NebulaGraph tells edges of one type between the same vertices apart by
    rank alone, so such an edge takes a signed 64-bit digest of those values,
    rendered as :func:`vertex_key` renders identities. ``None`` (rank 0) when
    there are none.
    """
    if not identity_fields:
        return None
    digest = hashlib.sha256(vertex_key(props, identity_fields).encode()).digest()
    return int.from_bytes(digest[:8], "big", signed=True)

escape_nebula_string(value)

Escape a string value for safe embedding in nGQL / GQL literals.

Source code in graflo/db/nebula/util.py

def escape_nebula_string(value: str) -> str:
    """Escape a string value for safe embedding in nGQL / GQL literals."""
    return value.replace("\\", "\\\\").replace('"', '\\"')

field_type_from_nebula(declared)

Map a DESCRIBE TAG / DESCRIBE EDGE type back to a FieldType.

Returns None for a type this mapping does not cover, so the caller can leave the field untyped rather than assert a wrong one.

Source code in graflo/db/nebula/util.py

def field_type_from_nebula(declared: str | None) -> FieldType | None:
    """Map a ``DESCRIBE TAG`` / ``DESCRIBE EDGE`` type back to a ``FieldType``.

    Returns ``None`` for a type this mapping does not cover, so the caller can
    leave the field untyped rather than assert a wrong one.
    """
    if not declared:
        return None
    # `fixed_string(32)` and friends carry a length the graflo model does not model.
    base = declared.strip().lower().split("(", 1)[0].strip()
    if base == "fixed_string":
        return FieldType.STRING
    return NEBULA_TYPE_TO_FIELD_TYPE.get(base)

fitted_vid_length(schema, string_length)

Bytes a FIXED_STRING VID needs for every vertex of schema.

A VID is tag::v1::v2… (:func:make_vid); a value whose type bounds its text takes that bound, any other string_length. Rounded up to 8 bytes. Without a schema: one string value under a tag of up to 64 bytes.

Source code in graflo/db/nebula/util.py

def fitted_vid_length(schema: Schema | None, string_length: int) -> int:
    """Bytes a ``FIXED_STRING`` VID needs for every vertex of *schema*.

    A VID is ``tag::v1::v2…`` (:func:`make_vid`); a value whose type bounds its
    text takes that bound, any other *string_length*. Rounded up to 8 bytes.
    Without a schema: one string value under a tag of up to 64 bytes.
    """

    def value_length(ft: FieldType | None) -> int:
        return string_length if ft is None else _VALUE_LENGTH.get(ft, string_length)

    sep = len(VID_SEPARATOR)
    lengths = [_TAG_ALLOWANCE + sep + string_length]
    if schema is not None and schema.core_schema.vertex_config.vertex_set:
        logical = schema.core_schema.vertex_config
        db_aware = schema.resolve_db_aware(DBType.NEBULA).vertex_config
        lengths = []
        for vertex in logical.vertex_set:
            types = {f.name: f.type for f in logical.properties(vertex)}
            fields = logical.identity_fields(vertex) or [""]
            tag = len(db_aware.vertex_dbname(vertex).encode())
            lengths.append(tag + sum(sep + value_length(types.get(f)) for f in fields))
    return -(-max(lengths) // 8) * 8

is_nebula_string_field(field)

Whether field becomes a variable-length string column.

Index DDL must ask the same question the column DDL answered: Nebula rejects CREATE TAG INDEX on a string column given without a length. Testing field.type == FieldType.STRING is not that question — an untyped field, a DATETIME and a UUID all land on string too, and an index over any of them is rejected as Invalid param!.

Source code in graflo/db/nebula/util.py

def is_nebula_string_field(field: Field) -> bool:
    """Whether *field* becomes a variable-length ``string`` column.

    Index DDL must ask the same question the column DDL answered: Nebula
    rejects ``CREATE TAG INDEX`` on a string column given without a length.
    Testing ``field.type == FieldType.STRING`` is not that question — an
    untyped field, a ``DATETIME`` and a ``UUID`` all land on ``string`` too,
    and an index over any of them is rejected as ``Invalid param!``.
    """
    return nebula_type(field.type) == DEFAULT_NEBULA_TYPE

key_of_vid(vid)

:func:vertex_key of a VID: what follows the tag. A tag name has no ::.

Source code in graflo/db/nebula/util.py

def key_of_vid(vid: str) -> str:
    """:func:`vertex_key` of a VID: what follows the tag. A tag name has no ``::``."""
    return vid.split(VID_SEPARATOR, 1)[1] if VID_SEPARATOR in vid else vid

make_vid(tag, doc, match_keys)

The VID of doc as a vertex of tag, keyed on match_keys.

Source code in graflo/db/nebula/util.py

def make_vid(tag: str, doc: dict[str, Any], match_keys: Sequence[str]) -> str:
    """The VID of *doc* as a vertex of *tag*, keyed on *match_keys*."""
    return tag_vid(tag, vertex_key(doc, match_keys))

nebula_type(ft)

Map a scalar graflo FieldType to the corresponding NebulaGraph type name.

LIST is not storable as a Nebula property — use :func:nebula_type_for_field.

Source code in graflo/db/nebula/util.py

def nebula_type(ft: FieldType | None) -> str:
    """Map a scalar graflo ``FieldType`` to the corresponding NebulaGraph type name.

    ``LIST`` is not storable as a Nebula property — use :func:`nebula_type_for_field`.
    """
    if ft is None:
        return DEFAULT_NEBULA_TYPE
    if is_list_field_type(ft):
        raise UnsupportedFieldTypeError(
            "LIST cannot be stored as a NebulaGraph vertex/edge property "
            "(composite types are query-only). No soft conversion to string/JSON."
        )
    return FIELD_TYPE_TO_NEBULA.get(ft, DEFAULT_NEBULA_TYPE)

nebula_type_for_field(field)

Map a Field to a NebulaGraph DDL type, raising for unsupported types.

Source code in graflo/db/nebula/util.py

def nebula_type_for_field(field: Field) -> str:
    """Map a ``Field`` to a NebulaGraph DDL type, raising for unsupported types."""
    assert_field_type_supported(DBType.NEBULA, field)
    return nebula_type(field.type)

render_filters_cypher(filters, doc_name)

Render a FilterExpression as a Cypher WHERE clause (without the keyword).

Source code in graflo/db/nebula/util.py

def render_filters_cypher(
    filters: list | dict | FilterExpression | None,
    doc_name: str,
) -> str:
    """Render a ``FilterExpression`` as a Cypher ``WHERE`` clause (without the keyword)."""
    if filters is None:
        return ""
    if not isinstance(filters, FilterExpression):
        ff = FilterExpression.from_dict(filters)
    else:
        ff = filters
    return str(ff(doc_name=doc_name, kind=ExpressionFlavor.CYPHER))

render_filters_ngql(filters, doc_name)

Render a FilterExpression as an nGQL WHERE clause (without the keyword).

Source code in graflo/db/nebula/util.py

def render_filters_ngql(
    filters: list | dict | FilterExpression | None,
    doc_name: str,
) -> str:
    """Render a ``FilterExpression`` as an nGQL ``WHERE`` clause (without the keyword)."""
    if filters is None:
        return ""
    if not isinstance(filters, FilterExpression):
        ff = FilterExpression.from_dict(filters)
    else:
        ff = filters
    return str(ff(doc_name=doc_name, kind=ExpressionFlavor.NGQL))

serialize_nebula_value(value)

Serialise a Python value into an nGQL literal string.

Source code in graflo/db/nebula/util.py

def serialize_nebula_value(value: Any) -> str:
    """Serialise a Python value into an nGQL literal string."""
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, Decimal):
        return str(float(value))
    if isinstance(value, datetime):
        return f'"{value.isoformat()}"'
    if isinstance(value, date):
        return f'"{value.isoformat()}"'
    if isinstance(value, dt_time):
        return f'"{value.isoformat()}"'
    if isinstance(value, (list, dict)):
        return f'"{escape_nebula_string(json.dumps(value, default=str))}"'
    return f'"{escape_nebula_string(str(value))}"'

tag_vid(tag, key)

The VID of the vertex of tag whose :func:vertex_key is key.

The tag leads the VID because a VID is space-wide: without it, vertices of two tags with the same identity values would be one vertex carrying both tags, and would share their edges.

Source code in graflo/db/nebula/util.py

def tag_vid(tag: str, key: str) -> str:
    """The VID of the vertex of *tag* whose :func:`vertex_key` is *key*.

    The tag leads the VID because a VID is space-wide: without it, vertices of
    two tags with the same identity values would be one vertex carrying both
    tags, and would share their edges.
    """
    return f"{tag}{VID_SEPARATOR}{key}"

vertex_key(doc, match_keys)

A vertex's identity values joined with ::: its VID without the tag.

Source code in graflo/db/nebula/util.py

def vertex_key(doc: dict[str, Any], match_keys: Sequence[str]) -> str:
    """A vertex's identity values joined with ``::``: its VID without the tag."""
    return VID_SEPARATOR.join(str(doc.get(k, "")) for k in match_keys)

wait_for_schema_propagation(adapter, check_statement, *, max_retries=30, interval=1.0)

Poll check_statement until it succeeds or retries are exhausted.

NebulaGraph propagates schema changes asynchronously across the cluster. After CREATE SPACE / CREATE TAG / CREATE EDGE, subsequent statements may fail until propagation completes (typically within two heartbeat cycles, ~20 s for default settings).

Source code in graflo/db/nebula/util.py

def wait_for_schema_propagation(
    adapter: NebulaClientAdapter,
    check_statement: str,
    *,
    max_retries: int = 30,
    interval: float = 1.0,
) -> None:
    """Poll *check_statement* until it succeeds or retries are exhausted.

    NebulaGraph propagates schema changes asynchronously across the cluster.
    After ``CREATE SPACE`` / ``CREATE TAG`` / ``CREATE EDGE``, subsequent
    statements may fail until propagation completes (typically within two
    heartbeat cycles, ~20 s for default settings).
    """
    for attempt in range(max_retries):
        try:
            adapter.execute(check_statement)
            return
        except Exception:
            if attempt == max_retries - 1:
                raise
            logger.debug(
                "Schema not yet propagated (attempt %d/%d), retrying in %.1fs …",
                attempt + 1,
                max_retries,
                interval,
            )
            time.sleep(interval)

wait_for_space_ready(adapter, space_name, *, max_retries=30, interval=1.0)

Wait until SE `space_name succeeds.

Source code in graflo/db/nebula/util.py

def wait_for_space_ready(
    adapter: NebulaClientAdapter,
    space_name: str,
    *,
    max_retries: int = 30,
    interval: float = 1.0,
) -> None:
    """Wait until ``USE `space_name``` succeeds."""
    wait_for_schema_propagation(
        adapter,
        f"USE `{space_name}`",
        max_retries=max_retries,
        interval=interval,
    )