python: expose observability and projected query APIs

python/README.md

@@ -161,6 +161,76 @@ value.pushType = "audioCall"
 Set metadata before publishing the value. Numeric values outside the native
 integer range raise `OverflowError`.
 
+## Observability
+
+`getNodeInfo()` returns a read-only `NodeInfo` snapshot with node identifiers,
+IPv4/IPv6 routing statistics, bound ports, pending-operation count, and local
+and remote storage counters. `getNodesStats(socket.AF_INET)` and
+`getNodesStats(socket.AF_INET6)` return `NodeStats` snapshots. Snapshots remain
+valid after the runner stops. `getPublicAddress()` returns known public
+`SockAddr` objects; pass an address family to restrict the results, or use
+`getPublicAddressStr()` for strings. These lists may be empty before discovery.
+`getNodeMessageStats(incoming=True)` selects incoming message counters.
+
+```python
+info = node.getNodeInfo()
+print(info.node_id, info.bound4, info.ipv4.good_nodes, info.storage_size)
+print(node.getPublicAddressStr())
+
+node.setOnStatusChanged(lambda ipv4, ipv6: print(ipv4.name, ipv6.name))
+node.setOnStatusChanged(None)
+```
+
+The status callback receives two `NodeStatus` values: `Disconnected`,
+`Connecting`, or `Connected`. Registration replaces the previous callback and
+persists across restarts; `None` disables it. Core callbacks run on the DHT
+thread while native locks are held: do not call snapshot getters, blocking
+operations, or `join()` from them. The asyncio wrapper schedules status
+callbacks on its event-loop thread, where snapshot getters are safe to use.
+
+## Queries And Filtered Listeners
+
+`query()` retrieves metadata rather than complete value payloads. Each row is
+a dictionary keyed by `ValueField`. Network projections support `Id`,
+`ValueType`, `OwnerPk`, and `UserType`; values are integers, an `InfoHash` owner
+fingerprint, or a UTF-8 string respectively. Explicit sequence-number
+projections are not supported by the native wire format.
+
+```python
+selection = dht.Select().field(dht.ValueField.Id).field(dht.ValueField.UserType)
+query = dht.Query()
+query.buildFrom(selection, dht.Where().userType("record"))
+rows = node.query(key, query=query)
+
+query = dht.Query("SELECT id,user_type WHERE user_type=record")
+node.query(key, query_cb=lambda row: print(row) or True,
+           done_cb=lambda ok, nodes: print(ok), query=query)
+```
+
+Without an explicit query, all four supported metadata fields are selected.
+Supplying `query_cb` makes the core call nonblocking; returning `False` stops
+delivery. The blocking core API returns the collected rows, as `get()` does.
+The asyncio API offers `query()` as a context-managed async iterator and
+`await queryAll()` as a collector; these raise `RuntimeError` on failure.
+
+```python
+rows = await node.queryAll(key, query=query)
+with node.query(key, query=query) as results:
+    async for row in results:
+        print(row[dht.ValueField.Id])
+```
+
+Both core and asyncio `listen()` accept optional `where` and `filter` arguments.
+`Where` is sent to the DHT to narrow the subscription; the Python predicate
+runs locally. Existing two-argument core listeners are unchanged.
+
+```python
+token = node.listen(key, lambda value, expired: True,
+                    where=dht.Where().userType("record"),
+                    filter=lambda value: len(value.data) < 1024)
+node.cancelListen(token)
+```
+
 ## Documentation
 
 For more information, examples and documentation, or to repport issues, visit: https://opendht.net

python/opendht.pyx

@@ -17,6 +17,7 @@ from libcpp.memory cimport shared_ptr, unique_ptr
 from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
 from cpython cimport ref
 from datetime import timedelta
+from enum import IntEnum as _IntEnum
 from operator import index as _index
 
 cimport opendht_cpp as cpp
@@ -24,6 +25,8 @@ cimport opendht_cpp as cpp
 ctypedef cpp.Pht* _PhtPtr
 
 import threading
+import socket as _socket
+import weakref as _weakref
 
 cdef inline void _initialize_gnutls():
     cdef int err = cpp.gnutls_global_init()
@@ -42,6 +45,58 @@ _initialize_gnutls()
 def version() -> str:
     return cpp.version().decode()
 
+class NodeStatus(_IntEnum):
+    Disconnected = 0
+    Connecting = 1
+    Connected = 2
+
+class ValueField(_IntEnum):
+    Id = 1
+    ValueType = 2
+    OwnerPk = 3
+    SeqNum = 4
+    UserType = 5
+
+class _StatusCallbackState:
+    __slots__ = ("callback", "__weakref__")
+    def __init__(self):
+        self.callback = None
+
+cdef inline void status_callback(int ipv4, int ipv6, void* user_data) noexcept with gil:
+    state = (<object>user_data)()
+    if state is None:
+        return
+    callback = state.callback
+    if callback is not None:
+        callback(NodeStatus(ipv4), NodeStatus(ipv6))
+
+cdef inline bool query_callback(shared_ptr[cpp.FieldValueIndex] value, void* user_data) noexcept with gil:
+    cdef cpp.Blob blob
+    row = {}
+    for entry in value.get().index:
+        field = ValueField(<int>entry.first)
+        if field == ValueField.OwnerPk:
+            owner = InfoHash()
+            owner._infohash = entry.second.getHash()
+            row[field] = owner
+        elif field == ValueField.UserType:
+            blob = entry.second.getBlob()
+            row[field] = (<char*>blob.data())[:blob.size()].decode()
+        else:
+            row[field] = entry.second.getInt()
+    return (<object>user_data)['query'](row)
+
+cdef inline void query_done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void* user_data) noexcept with gil:
+    callback = (<object>user_data)['done']
+    if callback is not None:
+        entries = []
+        for node in deref(nodes):
+            entry = NodeEntry()
+            entry._v.first = node.get().getId()
+            entry._v.second = node
+            entries.append(entry)
+        callback(done, entries)
+
 cdef inline void lookup_callback(cpp.vector[cpp.shared_ptr[cpp.IndexValue]]* values, cpp.Prefix* p, void *user_data) noexcept with gil:
     cbs = <object>user_data
     if 'lookup' in cbs and cbs['lookup']:
@@ -241,6 +296,78 @@ cdef class Node(_WithID):
             raise ValueError("Node is not initialized")
         return self._node.get().isExpired()
 
+cdef class NodeStats(object):
+    cdef cpp.NodeStats _stats
+    @property
+    def good_nodes(self):
+        return self._stats.good_nodes
+    @property
+    def dubious_nodes(self):
+        return self._stats.dubious_nodes
+    @property
+    def cached_nodes(self):
+        return self._stats.cached_nodes
+    @property
+    def incoming_nodes(self):
+        return self._stats.incoming_nodes
+    @property
+    def table_depth(self):
+        return self._stats.table_depth
+    @property
+    def searches(self):
+        return self._stats.searches
+    @property
+    def node_cache_size(self):
+        return self._stats.node_cache_size
+    def getKnownNodes(self):
+        return self._stats.getKnownNodes()
+    def getNetworkSizeEstimation(self):
+        return self._stats.getNetworkSizeEstimation()
+
+cdef class NodeInfo(object):
+    cdef cpp.NodeInfo _info
+    @property
+    def id(self):
+        result = InfoHash()
+        result._infohash = self._info.id
+        return result
+    @property
+    def node_id(self):
+        result = InfoHash()
+        result._infohash = self._info.node_id
+        return result
+    @property
+    def ipv4(self):
+        result = NodeStats()
+        result._stats = self._info.ipv4
+        return result
+    @property
+    def ipv6(self):
+        result = NodeStats()
+        result._stats = self._info.ipv6
+        return result
+    @property
+    def ongoing_ops(self):
+        return self._info.ongoing_ops
+    @property
+    def storage_values(self):
+        return self._info.storage_values
+    @property
+    def storage_size(self):
+        return self._info.storage_size
+    @property
+    def local_storage_values(self):
+        return self._info.local_storage_values
+    @property
+    def local_storage_size(self):
+        return self._info.local_storage_size
+    @property
+    def bound4(self):
+        return self._info.bound4
+    @property
+    def bound6(self):
+        return self._info.bound6
+
 cdef class NodeEntry(_WithID):
     cdef cpp.pair[cpp.InfoHash, shared_ptr[cpp.Node]] _v
     def getId(self):
@@ -274,8 +401,11 @@ cdef class Select(object):
         return self._select.toString().decode()
     def isSatisfiedBy(self, Select os):
         return self._select.isSatisfiedBy(os._select)
-    def field(self, int field):
-        self._select.field(<cpp.Field> field)
+    def field(self, field):
+        cdef int field_id = ValueField(_index(field))
+        if field_id == ValueField.SeqNum:
+            raise ValueError("The native wire format does not support sequence-number projections")
+        self._select.field(<cpp.Field>field_id)
         return self
 
 cdef class Where(object):
@@ -750,8 +880,10 @@ cdef class DhtConfig(object):
 cdef class DhtRunner(_WithID):
     cdef cpp.shared_ptr[cpp.DhtRunner] thisptr
     cdef map[_PhtPtr, shared_ptr[cpp.Pht]] _indexes
+    cdef object _status_callback
     def __cinit__(self):
         self.thisptr.reset(new cpp.DhtRunner())
+        self._status_callback = _StatusCallbackState()
     def __dealloc__(self):
         with nogil:
             if self.thisptr:
@@ -761,19 +893,51 @@ cdef class DhtRunner(_WithID):
     def getId(self) -> InfoHash:
         h = InfoHash()
         if self.thisptr:
-            h._infohash = self.thisptr.get().getId()
+            with nogil:
+                h._infohash = self.thisptr.get().getId()
         return h
     def getPublicKey(self) -> PublicKey:
         if not self.thisptr:
             raise ValueError("DhtRunner is not initialized")
         pk = PublicKey()
-        pk._key = self.thisptr.get().getPublicKey()
+        with nogil:
+            pk._key = self.thisptr.get().getPublicKey()
         return pk
     def getNodeId(self) -> InfoHash:
         h = InfoHash()
         if self.thisptr:
-            h._infohash = self.thisptr.get().getNodeId()
+            with nogil:
+                h._infohash = self.thisptr.get().getNodeId()
         return h
+    def getNodeInfo(self) -> NodeInfo:
+        result = NodeInfo()
+        with nogil:
+            result._info = self.thisptr.get().getNodeInfo()
+        return result
+    def getNodesStats(self, af=_socket.AF_INET) -> NodeStats:
+        if af not in (_socket.AF_INET, _socket.AF_INET6):
+            raise ValueError("af must be AF_INET or AF_INET6")
+        info = self.getNodeInfo()
+        return info.ipv4 if af == _socket.AF_INET else info.ipv6
+    def getPublicAddress(self, cpp.sa_family_t af=0):
+        if af not in (0, _socket.AF_INET, _socket.AF_INET6):
+            raise ValueError("af must be AF_UNSPEC, AF_INET or AF_INET6")
+        cdef cpp.vector[cpp.SockAddr] addresses
+        with nogil:
+            addresses = self.thisptr.get().getPublicAddress(af)
+        result = []
+        for address in addresses:
+            value = SockAddr()
+            value._addr = address
+            result.append(value)
+        return result
+    def getPublicAddressStr(self, cpp.sa_family_t af=0):
+        return [str(address) for address in self.getPublicAddress(af)]
+    def setOnStatusChanged(self, callback=None):
+        """Replace the DHT-thread status callback, or disable it with None."""
+        if callback is not None and not callable(callback):
+            raise TypeError("callback must be callable or None")
+        self._status_callback.callback = callback
     def ping(self, SockAddr addr, done_cb=None):
         if done_cb:
             cb_obj = {'done':done_cb}
@@ -808,9 +972,12 @@ cdef class DhtRunner(_WithID):
         if ipv4 or ipv6:
             bind4 = ipv4.encode() if ipv4 else b''
             bind6 = ipv6.encode() if ipv6 else b''
-            self.thisptr.get().run(bind4, bind6, str(port).encode(), run_config)
+            cpp.runPythonRunnerAddress(deref(self.thisptr.get()), bind4, bind6,
+                                       str(port).encode(), run_config,
+                                       status_callback, _weakref.ref(self._status_callback))
         else:
-            self.thisptr.get().run(port, run_config)
+            cpp.runPythonRunner(deref(self.thisptr.get()), port, run_config,
+                                status_callback, _weakref.ref(self._status_callback))
     def join(self):
         with nogil:
             self.thisptr.get().join()
@@ -831,17 +998,29 @@ cdef class DhtRunner(_WithID):
         return self.thisptr.get().isRunning()
     def getBound(self, cpp.sa_family_t af = 0):
         s = SockAddr()
-        s._addr = self.thisptr.get().getBound(af)
+        with nogil:
+            s._addr = self.thisptr.get().getBound(af)
         return s
     def getStorageLog(self):
-        return self.thisptr.get().getStorageLog().decode()
+        cdef cpp.string result
+        with nogil:
+            result = self.thisptr.get().getStorageLog()
+        return result.decode()
     def getRoutingTablesLog(self, cpp.sa_family_t af):
-        return self.thisptr.get().getRoutingTablesLog(af).decode()
+        cdef cpp.string result
+        with nogil:
+            result = self.thisptr.get().getRoutingTablesLog(af)
+        return result.decode()
     def getSearchesLog(self, cpp.sa_family_t af):
-        return self.thisptr.get().getSearchesLog(af).decode()
-    def getNodeMessageStats(self):
+        cdef cpp.string result
+        with nogil:
+            result = self.thisptr.get().getSearchesLog(af)
+        return result.decode()
+    def getNodeMessageStats(self, bool incoming=False):
         stats = []
-        cdef cpp.vector[unsigned] res = self.thisptr.get().getNodeMessageStats(False)
+        cdef cpp.vector[unsigned] res
+        with nogil:
+            res = self.thisptr.get().getNodeMessageStats(incoming)
         for n in res:
             stats.append(n)
         return stats
@@ -884,6 +1063,25 @@ cdef class DhtRunner(_WithID):
                     lock.wait()
             return res
 
+    def query(self, InfoHash key not None, query_cb=None, done_cb=None, Query query=None):
+        """Return projected metadata rows, or stream them through query_cb."""
+        if query is None:
+            query = Query("SELECT id,value_type,owner_pk,user_type")
+        if query_cb is None:
+            completed = threading.Event()
+            rows = []
+            def on_row(row):
+                rows.append(row)
+                return True
+            self.query(key, on_row, lambda ok, nodes: completed.set(), query)
+            completed.wait()
+            return rows
+        if not callable(query_cb) or (done_cb is not None and not callable(done_cb)):
+            raise TypeError("query callbacks must be callable")
+        callbacks = {'query': query_cb, 'done': done_cb}
+        cpp.queryPython(deref(self.thisptr.get()), key._infohash, query_callback,
+                        query_done_callback, callbacks, query._query)
+
     def put(self, InfoHash key, Value val, done_cb=None, permanent=False):
         """Publish a new value on the DHT at key.
 
@@ -976,11 +1174,16 @@ cdef class DhtRunner(_WithID):
     def cancelPut(self, InfoHash key, Value val):
         self.thisptr.get().cancelPut(key._infohash, val._value)
 
-    def listen(self, InfoHash key, value_cb):
+    def listen(self, InfoHash key not None, value_cb, filter=None, Where where=None):
+        if not callable(value_cb) or (filter is not None and not callable(filter)):
+            raise TypeError("listener callbacks must be callable")
+        if where is None:
+            where = Where()
         t = ListenToken()
         t._h = key._infohash
-        cb_obj = {'valcb':value_cb}
-        t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj)).share()
+        cb_obj = {'valcb':value_cb, 'filter':filter}
+        t._t = self.thisptr.get().listen(t._h, cpp.bindPythonValueCb(value_callback, cb_obj),
+                                       cpp.nullptr, where._where).share()
         return t
     def cancelListen(self, ListenToken token):
         if token is None:

python/opendht/aio.py

@@ -3,12 +3,15 @@
 from __future__ import annotations
 
 import asyncio
-from typing import Any, Callable, Iterable, List, Optional, Tuple
+from typing import Any, Callable, Generic, Iterable, List, Optional, Tuple, TypeVar
 
 from . import _core as _dht
 
 # Re-export core types for convenience
 from ._core import InfoHash, Value, ValueType, Where, SockAddr, PublicKey, PkId, DhtConfig, Identity
+from ._core import NodeInfo, NodeStats, NodeStatus, Query, Select, ValueField
+
+_Result = TypeVar("_Result")
 
 
 def _set_result_if_pending(future: asyncio.Future[Any], result: Any) -> None:
@@ -24,7 +27,11 @@ class _Listener:
 
     __slots__ = ("_runner", "_token", "_loop", "_queue", "_closed")
 
-    def __init__(self, runner: DhtRunner, key: InfoHash) -> None:
+    def __init__(
+        self, runner: DhtRunner, key: InfoHash,
+        where: Optional[Where] = None,
+        filter: Optional[Callable[[Value], bool]] = None,
+    ) -> None:
         self._runner = runner
         self._loop = runner._loop
         self._queue: asyncio.Queue[Tuple[Value, bool]] = asyncio.Queue()
@@ -36,7 +43,7 @@ class _Listener:
             self._loop.call_soon_threadsafe(self._queue.put_nowait, (v, expired))
             return True  # continue listening
 
-        self._token = self._runner._dht.listen(key, _on_value)
+        self._token = self._runner._dht.listen(key, _on_value, filter, where)
 
     def __enter__(self) -> _Listener:
         return self
@@ -67,10 +74,10 @@ class _Listener:
             self._queue.put_nowait(None)
 
 
-class _GetStream:
-    """Async iterator around DHT get callback.
+class _GetStream(Generic[_Result]):
+    """Async iterator around a DHT get or query callback.
 
-    Yields Value instances as they are found, then ends.
+    Yields results as they are found, then ends.
     Raises on failure.
     """
     __slots__ = ("_runner", "_loop", "_queue", "_closed", "_error")
@@ -81,14 +88,15 @@ class _GetStream:
         key: InfoHash,
         where: Optional[Where],
         filter: Optional[Callable[[Value], bool]],
+        query: Optional[Query] = None,
     ) -> None:
         self._runner = runner
         self._loop = runner._loop
-        self._queue: asyncio.Queue[Optional[Value]] = asyncio.Queue()
+        self._queue: asyncio.Queue[Optional[_Result]] = asyncio.Queue()
         self._closed = False
         self._error: Optional[BaseException] = None
 
-        def _on_value(v: Value) -> bool:
+        def _on_value(v: _Result) -> bool:
             # Called on DHT thread
             if self._closed:
                 return False
@@ -100,21 +108,24 @@ class _GetStream:
             if self._closed:
                 return
             if not ok:
-                self._error = RuntimeError("DHT get failed")
+                self._error = RuntimeError("DHT query failed" if query is not None else "DHT get failed")
             self._loop.call_soon_threadsafe(self._queue.put_nowait, None)
 
-        self._runner._dht.get(key, _on_value, _on_done, filter, where)
+        if query is None:
+            self._runner._dht.get(key, _on_value, _on_done, filter, where)
+        else:
+            self._runner._dht.query(key, _on_value, _on_done, query)
 
-    def __enter__(self) -> _GetStream:
+    def __enter__(self) -> _GetStream[_Result]:
         return self
 
     def __exit__(self, exc_type, exc, tb) -> None:
         self.cancel()
 
-    def __aiter__(self) -> _GetStream:
+    def __aiter__(self) -> _GetStream[_Result]:
         return self
 
-    async def __anext__(self) -> Value:
+    async def __anext__(self) -> _Result:
         if self._closed:
             raise StopAsyncIteration
         item = await self._queue.get()
@@ -137,7 +148,7 @@ class DhtRunner:
     """Async wrapper around opendht.DhtRunner.
     """
 
-    __slots__ = ("_dht", "_loop", "_config", "_bootstrap")
+    __slots__ = ("_dht", "_loop", "_config", "_bootstrap", "_status_generation")
 
     def __init__(
         self,
@@ -154,6 +165,7 @@ class DhtRunner:
         self._dht = core or _dht.DhtRunner()
         self._config = config
         self._bootstrap = list(bootstrap) if bootstrap is not None else None
+        self._status_generation = [0]
 
     def run(
         self,
@@ -173,6 +185,29 @@ class DhtRunner:
     def bootstrap(self, host: str, port: Optional[str] = None) -> None:
         self._dht.bootstrap(host, port)
 
+    def setOnStatusChanged(
+        self, callback: Optional[Callable[[NodeStatus, NodeStatus], None]] = None,
+    ) -> None:
+        if callback is not None and not callable(callback):
+            raise TypeError("callback must be callable or None")
+        state = self._status_generation
+        state[0] += 1
+        if callback is None:
+            self._dht.setOnStatusChanged(None)
+            return
+        generation = state[0]
+        loop = self._loop
+
+        def _deliver(ipv4: NodeStatus, ipv6: NodeStatus) -> None:
+            if state[0] == generation:
+                callback(ipv4, ipv6)
+
+        def _on_status(ipv4: NodeStatus, ipv6: NodeStatus) -> None:
+            if not loop.is_closed():
+                loop.call_soon_threadsafe(_deliver, ipv4, ipv6)
+
+        self._dht.setOnStatusChanged(_on_status)
+
     async def ping(self, addr: SockAddr) -> bool:
         fut: asyncio.Future[bool] = self._loop.create_future()
         self._dht.ping(addr, lambda ok: self._loop.call_soon_threadsafe(_set_result_if_pending, fut, bool(ok)))
@@ -184,9 +219,25 @@ class DhtRunner:
         *,
         where: Optional[Where] = None,
         filter: Optional[Callable[[Value], bool]] = None,
-    ):
+    ) -> _GetStream[Value]:
         # Return a streaming iterator of values; it ends on completion and raises on failure.
-        return _GetStream(self, key, where, filter)
+        return _GetStream[Value](self, key, where, filter)
+
+    def query(
+        self, key: InfoHash, *, query: Optional[Query] = None,
+    ) -> _GetStream[dict[ValueField, Any]]:
+        if query is None:
+            query = Query("SELECT id,value_type,owner_pk,user_type")
+        return _GetStream[dict[ValueField, Any]](self, key, None, None, query)
+
+    async def queryAll(
+        self, key: InfoHash, *, query: Optional[Query] = None,
+    ) -> List[dict[ValueField, Any]]:
+        rows: List[dict[ValueField, Any]] = []
+        with self.query(key, query=query) as stream:
+            async for row in stream:
+                rows.append(row)
+        return rows
 
     async def getAll(
         self,
@@ -255,8 +306,11 @@ class DhtRunner:
         self._dht.putEncrypted(key, val, to, _done, permanent)
         return await fut
 
-    def listen(self, key: InfoHash) -> _Listener:
-        return _Listener(self, key)
+    def listen(
+        self, key: InfoHash, *, where: Optional[Where] = None,
+        filter: Optional[Callable[[Value], bool]] = None,
+    ) -> _Listener:
+        return _Listener(self, key, where, filter)
 
     async def cancelListen(self, listener: _Listener) -> None:
         listener.cancel()

python/opendht_cpp.pxd

@@ -155,6 +155,7 @@ cdef extern from "opendht/value.h" namespace "dht::Value":
 cdef extern from "opendht/value.h" namespace "dht::Value::Field":
     cdef Field None
     cdef Field Id
+    cdef Field ValueTypeField "dht::Value::Field::ValueType"
     cdef Field OwnerPk
     cdef Field SeqNum
     cdef Field UserType
@@ -208,6 +209,14 @@ cdef extern from "opendht/value.h" namespace "dht":
         Where& userType(string user_type)
         string toString() const
 
+    cdef cppclass FieldValue:
+        uint64_t getInt() const
+        InfoHash getHash() const
+        Blob getBlob() const
+
+    cdef cppclass FieldValueIndex:
+        map[Field, FieldValue] index
+
 cdef extern from "opendht/node.h" namespace "dht":
     cdef cppclass Node:
         Node() except +
@@ -216,6 +225,30 @@ cdef extern from "opendht/node.h" namespace "dht":
         bool isExpired() const
 
 cdef extern from "opendht/callbacks.h" namespace "dht":
+    cdef cppclass NodeStats:
+        unsigned good_nodes
+        unsigned dubious_nodes
+        unsigned cached_nodes
+        unsigned incoming_nodes
+        unsigned table_depth
+        unsigned searches
+        unsigned node_cache_size
+        unsigned getKnownNodes() const
+        unsigned long getNetworkSizeEstimation() const
+
+    cdef cppclass NodeInfo:
+        InfoHash id
+        InfoHash node_id
+        NodeStats ipv4
+        NodeStats ipv6
+        size_t ongoing_ops
+        size_t storage_values
+        size_t storage_size
+        size_t local_storage_values
+        size_t local_storage_size
+        in_port_t bound4
+        in_port_t bound6
+
     ctypedef void (*ShutdownCallbackRaw)(void *user_data)
     ctypedef bool (*GetCallbackRaw)(shared_ptr[Value] values, void *user_data)
     ctypedef bool (*ValueCallbackRaw)(shared_ptr[Value] values, bool expired, void *user_data)
@@ -254,17 +287,22 @@ cdef extern from "opendht/callbacks.h" namespace "dht":
 cdef extern from *:
     """
     #include <Python.h>
-    #include <opendht/callbacks.h>
+    #include <opendht/dhtrunner.h>
 
-    static dht::ValueCallback opendht_bind_python_value_cb(
-        dht::ValueCallbackRaw callback, PyObject* data)
+    static std::shared_ptr<PyObject> opendht_python_owner(PyObject* data)
     {
         Py_INCREF(data);
-        auto owner = std::shared_ptr<PyObject>(data, [](PyObject* object) {
+        return std::shared_ptr<PyObject>(data, [](PyObject* object) {
             auto state = PyGILState_Ensure();
             Py_DECREF(object);
             PyGILState_Release(state);
         });
+    }
+
+    static dht::ValueCallback opendht_bind_python_value_cb(
+        dht::ValueCallbackRaw callback, PyObject* data)
+    {
+        auto owner = opendht_python_owner(data);
         return [callback, owner = std::move(owner)](
                    const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
             for (const auto& value : values)
@@ -273,6 +311,48 @@ cdef extern from *:
             return true;
         };
     }
+
+    static dht::DhtRunner::Context opendht_python_context(
+        void (*callback)(int, int, void*), PyObject* data)
+    {
+        dht::DhtRunner::Context context;
+        auto owner = opendht_python_owner(data);
+        context.statusChangedCallback = [callback, owner = std::move(owner)](
+            dht::NodeStatus ipv4, dht::NodeStatus ipv6) {
+            callback(static_cast<int>(ipv4), static_cast<int>(ipv6), owner.get());
+        };
+        return context;
+    }
+
+    static void opendht_python_run(dht::DhtRunner& runner, in_port_t port,
+        dht::DhtRunner::Config& config, void (*callback)(int, int, void*), PyObject* data)
+    {
+        runner.run(port, config, opendht_python_context(callback, data));
+    }
+
+    static void opendht_python_run(dht::DhtRunner& runner,
+        const char* ipv4, const char* ipv6, const char* port,
+        dht::DhtRunner::Config& config, void (*callback)(int, int, void*), PyObject* data)
+    {
+        runner.run(ipv4, ipv6, port, config, opendht_python_context(callback, data));
+    }
+
+    static void opendht_python_query(dht::DhtRunner& runner, const dht::InfoHash& key,
+        bool (*callback)(std::shared_ptr<dht::FieldValueIndex>, void*),
+        dht::DoneCallbackRaw done, PyObject* data, dht::Query query)
+    {
+        auto owner = opendht_python_owner(data);
+        runner.query(key,
+            [callback, owner](const std::vector<std::shared_ptr<dht::FieldValueIndex>>& rows) {
+                for (const auto& row : rows)
+                    if (!callback(row, owner.get()))
+                        return false;
+                return true;
+            },
+            [done, owner](bool ok, const std::vector<std::shared_ptr<dht::Node>>& nodes) {
+                done(ok, const_cast<std::vector<std::shared_ptr<dht::Node>>*>(&nodes), owner.get());
+            }, std::move(query));
+    }
     """
     ValueCallback bindPythonValueCb "opendht_bind_python_value_cb"(ValueCallbackRaw callback, object data) except +
 
@@ -298,9 +378,11 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
             SockAddr bind4
             SockAddr bind6
 
-        InfoHash getId() const
-        shared_ptr[PublicKey] getPublicKey() const
-        InfoHash getNodeId() const
+        InfoHash getId() except + nogil
+        shared_ptr[PublicKey] getPublicKey() except + nogil
+        InfoHash getNodeId() except + nogil
+        NodeInfo getNodeInfo() except + nogil
+        vector[SockAddr] getPublicAddress(sa_family_t af) except + nogil
         void bootstrap(const_char*, const_char*)
         void bootstrap(const SockAddr&, DoneCallbackSimple done_cb)
         void run(in_port_t, Config config) except +
@@ -309,22 +391,29 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         void shutdown(ShutdownCallback)
         void registerType(ValueType& value);
         bool isRunning()
-        SockAddr getBound(sa_family_t af) const
-        string getStorageLog() const
-        string getRoutingTablesLog(sa_family_t af) const
-        string getSearchesLog(sa_family_t af) const
+        SockAddr getBound(sa_family_t af) except + nogil
+        string getStorageLog() except + nogil
+        string getRoutingTablesLog(sa_family_t af) except + nogil
+        string getSearchesLog(sa_family_t af) except + nogil
         void get(InfoHash key, GetCallback get_cb, DoneCallback done_cb, nullptr_t f, Where w)
         void put(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb, time_point created, bool permanent)
         void putSigned(InfoHash key, shared_ptr[Value] val, DoneCallback done_cb, bool permanent) 
         void putEncrypted(InfoHash key, PkId to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent)
         void putEncrypted(InfoHash key, shared_ptr[PublicKey] to, shared_ptr[Value] val, DoneCallback done_cb, bool permanent)
         void cancelPut(InfoHash key, shared_ptr[Value] val)
-        ListenToken listen(InfoHash key, ValueCallback get_cb) except +
+        ListenToken listen(InfoHash key, ValueCallback get_cb, nullptr_t f, Where w) except +
         void cancelListen(InfoHash key, SharedListenToken token) except +
-        vector[unsigned] getNodeMessageStats(bool i)
+        vector[unsigned] getNodeMessageStats(bool i) except + nogil
 
 ctypedef DhtRunner.Config DhtRunnerConfig
 
+cdef extern from *:
+    ctypedef void (*StatusCallbackRaw)(int ipv4, int ipv6, void* data)
+    ctypedef bool (*QueryCallbackRaw)(shared_ptr[FieldValueIndex] row, void* data)
+    void runPythonRunner "opendht_python_run"(DhtRunner& runner, in_port_t port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except +
+    void runPythonRunnerAddress "opendht_python_run"(DhtRunner& runner, const_char* ipv4, const_char* ipv6, const_char* port, DhtRunnerConfig& config, StatusCallbackRaw callback, object data) except +
+    void queryPython "opendht_python_query"(DhtRunner& runner, InfoHash key, QueryCallbackRaw callback, DoneCallbackRaw done, object data, Query query) except +
+
 cdef extern from "opendht/log.h" namespace "dht::log":
     void enableLogging(DhtRunner& dht)
     void disableLogging(DhtRunner& dht)