Skip to content

graflo.db.nebula.adapter

Version-specific client adapters for NebulaGraph.

Provides a uniform execute / close / use_space interface over nebula3-python (v3.x, Thrift) and nebula5-python (v5.x, gRPC).

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

NebulaClientAdapter

Bases: ABC

Abstract adapter that hides driver differences.

Source code in graflo/db/nebula/adapter.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class NebulaClientAdapter(abc.ABC):
    """Abstract adapter that hides driver differences."""

    @abc.abstractmethod
    def connect(self, config: NebulaConfig) -> None: ...

    @abc.abstractmethod
    def execute(self, statement: str) -> NebulaResultSet: ...

    @abc.abstractmethod
    def close(self) -> None: ...

    @abc.abstractmethod
    def use_space(self, space_name: str) -> None: ...

Methods:

close() abstractmethod
Source code in graflo/db/nebula/adapter.py 
75
76
@abc.abstractmethod
def close(self) -> None: ...
connect(config) abstractmethod
Source code in graflo/db/nebula/adapter.py 
69
70
@abc.abstractmethod
def connect(self, config: NebulaConfig) -> None: ...
execute(statement) abstractmethod
Source code in graflo/db/nebula/adapter.py 
72
73
@abc.abstractmethod
def execute(self, statement: str) -> NebulaResultSet: ...
use_space(space_name) abstractmethod
Source code in graflo/db/nebula/adapter.py 
78
79
@abc.abstractmethod
def use_space(self, space_name: str) -> None: ...

NebulaResultSet

Thin wrapper around driver-specific result objects.

Normalises access so that NebulaConnection never has to know which driver is in use.

Source code in graflo/db/nebula/adapter.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class NebulaResultSet:
    """Thin wrapper around driver-specific result objects.

    Normalises access so that ``NebulaConnection`` never has to know which
    driver is in use.
    """

    def __init__(self, raw: Any, *, is_v3: bool = True):
        self._raw = raw
        self._is_v3 = is_v3

    @property
    def raw(self) -> Any:
        return self._raw

    def is_succeeded(self) -> bool:
        """Whether the statement succeeded.

        The two drivers report failure differently: v3 returns a status on the
        result, while v5 raises from :meth:`ResultSet.raise_on_error`, which
        :meth:`NebulaV5Adapter.execute` calls before wrapping. So a v5 result
        that reaches here has already been checked -- ``True`` is a fact about
        an error-free path, not an assumption that nothing failed.
        """
        if self._is_v3:
            return self._raw.is_succeeded()
        return True

    def error_msg(self) -> str:
        """Driver error message; empty on v5, where failures raise instead."""
        if self._is_v3:
            return self._raw.error_msg()
        return ""

    def column_values(self, col: str) -> list[Any]:
        if self._is_v3:
            return [v.cast() for v in self._raw.column_values(col)]
        return self._raw.as_primitive_by_column().get(col, [])

    def rows_as_dicts(self) -> list[dict[str, Any]]:
        """Return all rows as list of primitive-type dicts."""
        if self._is_v3:
            return self._raw.as_primitive()
        # nebula5-python exposes no as_primitive(); rows come back as an
        # iterator, so materialise it to match the v3 contract.
        return list(self._raw.as_primitive_by_row())

Attributes

raw property

Methods:

__init__(raw, *, is_v3=True)
Source code in graflo/db/nebula/adapter.py 
25
26
27
def __init__(self, raw: Any, *, is_v3: bool = True):
    self._raw = raw
    self._is_v3 = is_v3
column_values(col)
Source code in graflo/db/nebula/adapter.py 
52
53
54
55
def column_values(self, col: str) -> list[Any]:
    if self._is_v3:
        return [v.cast() for v in self._raw.column_values(col)]
    return self._raw.as_primitive_by_column().get(col, [])
error_msg()

Driver error message; empty on v5, where failures raise instead.

Source code in graflo/db/nebula/adapter.py 
46
47
48
49
50
def error_msg(self) -> str:
    """Driver error message; empty on v5, where failures raise instead."""
    if self._is_v3:
        return self._raw.error_msg()
    return ""
is_succeeded()

Whether the statement succeeded.

The two drivers report failure differently: v3 returns a status on the result, while v5 raises from :meth:ResultSet.raise_on_error, which :meth:NebulaV5Adapter.execute calls before wrapping. So a v5 result that reaches here has already been checked -- True is a fact about an error-free path, not an assumption that nothing failed.

Source code in graflo/db/nebula/adapter.py 
33
34
35
36
37
38
39
40
41
42
43
44
def is_succeeded(self) -> bool:
    """Whether the statement succeeded.

    The two drivers report failure differently: v3 returns a status on the
    result, while v5 raises from :meth:`ResultSet.raise_on_error`, which
    :meth:`NebulaV5Adapter.execute` calls before wrapping. So a v5 result
    that reaches here has already been checked -- ``True`` is a fact about
    an error-free path, not an assumption that nothing failed.
    """
    if self._is_v3:
        return self._raw.is_succeeded()
    return True
rows_as_dicts()

Return all rows as list of primitive-type dicts.

Source code in graflo/db/nebula/adapter.py 
57
58
59
60
61
62
63
def rows_as_dicts(self) -> list[dict[str, Any]]:
    """Return all rows as list of primitive-type dicts."""
    if self._is_v3:
        return self._raw.as_primitive()
    # nebula5-python exposes no as_primitive(); rows come back as an
    # iterator, so materialise it to match the v3 contract.
    return list(self._raw.as_primitive_by_row())

NebulaV3Adapter

Bases: NebulaClientAdapter

Adapter for nebula3-python (NebulaGraph 3.x, Thrift).

Source code in graflo/db/nebula/adapter.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class NebulaV3Adapter(NebulaClientAdapter):
    """Adapter for ``nebula3-python`` (NebulaGraph 3.x, Thrift)."""

    def __init__(self) -> None:
        self._pool: Any = None
        self._session: Any = None

    def connect(self, config: NebulaConfig) -> None:
        from nebula3.Config import Config as N3Config
        from nebula3.gclient.net import ConnectionPool

        hostname = config.hostname or "localhost"
        port = config.port or 9669
        username = config.username or "root"
        password = config.password or "nebula"

        n3_cfg = N3Config()
        n3_cfg.max_connection_pool_size = 10
        n3_cfg.timeout = int(config.request_timeout * 1000)

        self._pool = ConnectionPool()
        ok = self._pool.init([(hostname, port)], n3_cfg)
        if not ok:
            raise ConnectionError(
                f"Failed to connect to NebulaGraph at {hostname}:{port}"
            )

        self._session = self._pool.get_session(username, password)
        logger.info("Connected to NebulaGraph 3.x at %s:%s", hostname, port)

    def execute(self, statement: str) -> NebulaResultSet:
        if self._session is None:
            raise RuntimeError("Not connected")
        result = self._session.execute(statement)
        rs = NebulaResultSet(result, is_v3=True)
        if not rs.is_succeeded():
            raise RuntimeError(
                f"nGQL execution failed: {rs.error_msg()}\nStatement: {statement}"
            )
        return rs

    def use_space(self, space_name: str) -> None:
        self.execute(f"USE `{space_name}`")

    def close(self) -> None:
        if self._session is not None:
            self._session.release()
            self._session = None
        if self._pool is not None:
            self._pool.close()
            self._pool = None

Methods:

__init__()
Source code in graflo/db/nebula/adapter.py 
85
86
87
def __init__(self) -> None:
    self._pool: Any = None
    self._session: Any = None
close()
Source code in graflo/db/nebula/adapter.py 
126
127
128
129
130
131
132
def close(self) -> None:
    if self._session is not None:
        self._session.release()
        self._session = None
    if self._pool is not None:
        self._pool.close()
        self._pool = None
connect(config)
Source code in graflo/db/nebula/adapter.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def connect(self, config: NebulaConfig) -> None:
    from nebula3.Config import Config as N3Config
    from nebula3.gclient.net import ConnectionPool

    hostname = config.hostname or "localhost"
    port = config.port or 9669
    username = config.username or "root"
    password = config.password or "nebula"

    n3_cfg = N3Config()
    n3_cfg.max_connection_pool_size = 10
    n3_cfg.timeout = int(config.request_timeout * 1000)

    self._pool = ConnectionPool()
    ok = self._pool.init([(hostname, port)], n3_cfg)
    if not ok:
        raise ConnectionError(
            f"Failed to connect to NebulaGraph at {hostname}:{port}"
        )

    self._session = self._pool.get_session(username, password)
    logger.info("Connected to NebulaGraph 3.x at %s:%s", hostname, port)
execute(statement)
Source code in graflo/db/nebula/adapter.py 
112
113
114
115
116
117
118
119
120
121
def execute(self, statement: str) -> NebulaResultSet:
    if self._session is None:
        raise RuntimeError("Not connected")
    result = self._session.execute(statement)
    rs = NebulaResultSet(result, is_v3=True)
    if not rs.is_succeeded():
        raise RuntimeError(
            f"nGQL execution failed: {rs.error_msg()}\nStatement: {statement}"
        )
    return rs
use_space(space_name)
Source code in graflo/db/nebula/adapter.py 
123
124
def use_space(self, space_name: str) -> None:
    self.execute(f"USE `{space_name}`")

NebulaV5Adapter

Bases: NebulaClientAdapter

Adapter for nebula5-python (NebulaGraph 5.x, gRPC / ISO GQL).

.. warning:: Experimental and unverified. NebulaGraph 5.x ships no open-source server image, so this path is exercised by no test and no CI job. Most statement builders in :mod:graflo.db.nebula.query still emit nGQL rather than GQL, so DDL, writes and introspection are expected to fail here. Treat 5.x as unsupported until it can be run against a real instance; 3.x is the validated target.

Source code in graflo/db/nebula/adapter.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class NebulaV5Adapter(NebulaClientAdapter):
    """Adapter for ``nebula5-python`` (NebulaGraph 5.x, gRPC / ISO GQL).

    .. warning::
       **Experimental and unverified.** NebulaGraph 5.x ships no open-source
       server image, so this path is exercised by no test and no CI job. Most
       statement builders in :mod:`graflo.db.nebula.query` still emit nGQL
       rather than GQL, so DDL, writes and introspection are expected to fail
       here. Treat 5.x as unsupported until it can be run against a real
       instance; 3.x is the validated target.
    """

    def __init__(self) -> None:
        self._client: Any = None

    def connect(self, config: NebulaConfig) -> None:
        from nebulagraph_python.client import NebulaClient

        hostname = config.hostname or "localhost"
        port = config.port or 9669
        username = config.username or "root"
        password = config.password or "nebula"

        self._client = NebulaClient(
            addresses=f"{hostname}:{port}",
            user_name=username,
            password=password,
        )
        logger.warning(
            "Connected to NebulaGraph 5.x at %s:%s using the experimental v5 "
            "adapter. This path is unverified -- no open-source 5.x server "
            "image exists, so it is covered by no test, and most statement "
            "builders still emit nGQL rather than GQL. Use NebulaGraph 3.x for "
            "supported behaviour.",
            hostname,
            port,
        )

    def execute(self, statement: str) -> NebulaResultSet:
        if self._client is None:
            raise RuntimeError("Not connected")
        result = self._client.execute(statement)
        # Without this the driver's error status is never inspected and every
        # failed statement is reported as a success.
        result.raise_on_error()
        return NebulaResultSet(result, is_v3=False)

    def use_space(self, space_name: str) -> None:
        self.execute(f"USE `{space_name}`")

    def close(self) -> None:
        if self._client is not None:
            self._client.close()
            self._client = None

Methods:

__init__()
Source code in graflo/db/nebula/adapter.py 
147
148
def __init__(self) -> None:
    self._client: Any = None
close()
Source code in graflo/db/nebula/adapter.py 
185
186
187
188
def close(self) -> None:
    if self._client is not None:
        self._client.close()
        self._client = None
connect(config)
Source code in graflo/db/nebula/adapter.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def connect(self, config: NebulaConfig) -> None:
    from nebulagraph_python.client import NebulaClient

    hostname = config.hostname or "localhost"
    port = config.port or 9669
    username = config.username or "root"
    password = config.password or "nebula"

    self._client = NebulaClient(
        addresses=f"{hostname}:{port}",
        user_name=username,
        password=password,
    )
    logger.warning(
        "Connected to NebulaGraph 5.x at %s:%s using the experimental v5 "
        "adapter. This path is unverified -- no open-source 5.x server "
        "image exists, so it is covered by no test, and most statement "
        "builders still emit nGQL rather than GQL. Use NebulaGraph 3.x for "
        "supported behaviour.",
        hostname,
        port,
    )
execute(statement)
Source code in graflo/db/nebula/adapter.py 
173
174
175
176
177
178
179
180
def execute(self, statement: str) -> NebulaResultSet:
    if self._client is None:
        raise RuntimeError("Not connected")
    result = self._client.execute(statement)
    # Without this the driver's error status is never inspected and every
    # failed statement is reported as a success.
    result.raise_on_error()
    return NebulaResultSet(result, is_v3=False)
use_space(space_name)
Source code in graflo/db/nebula/adapter.py 
182
183
def use_space(self, space_name: str) -> None:
    self.execute(f"USE `{space_name}`")

Functions:

create_adapter(config)

Source code in graflo/db/nebula/adapter.py 
191
192
193
194
195
196
197
198
199
def create_adapter(config: NebulaConfig) -> NebulaClientAdapter:
    """Factory: instantiate and connect the correct adapter for *config.version*."""
    adapter: NebulaClientAdapter
    if config.is_v3:
        adapter = NebulaV3Adapter()
    else:
        adapter = NebulaV5Adapter()
    adapter.connect(config)
    return adapter