From 615534fa0f8dba839e53696be93a0d1888dbd154 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= <> Date: Tue, 2 Sep 2025 01:41:55 -0400 Subject: [PATCH] python: add initial asyncio implementation --- python/CMakeLists.txt | 4 +- python/opendht/__init__.py | 11 ++ python/opendht/aio.py | 336 +++++++++++++++++++++++++++++++++++++ python/setup.py.in | 8 +- 4 files changed, 355 insertions(+), 4 deletions(-) create mode 100644 python/opendht/__init__.py create mode 100644 python/opendht/aio.py diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt index 2651f62d..2ed1bdf9 100644 --- a/python/CMakeLists.txt +++ b/python/CMakeLists.txt @@ -9,7 +9,9 @@ find_package(Python3 REQUIRED COMPONENTS Interpreter) add_custom_target(python COMMAND ${Python3_EXECUTABLE} -m build . - DEPENDS opendht opendht_cpp.pxd opendht.pyx pyproject.toml) + DEPENDS opendht opendht_cpp.pxd opendht.pyx pyproject.toml + ${CURRENT_SOURCE_DIR}/opendht/__init__.py + ${CURRENT_SOURCE_DIR}/opendht/aio.py) install(CODE [[ execute_process( diff --git a/python/opendht/__init__.py b/python/opendht/__init__.py new file mode 100644 index 00000000..81c21ea9 --- /dev/null +++ b/python/opendht/__init__.py @@ -0,0 +1,11 @@ +""" +Python package facade for OpenDHT. + +This package re-exports the Cython-built core bindings from ``opendht._core`` +to preserve the public API (e.g., ``opendht.DhtRunner``), and also exposes a +pure-Python asyncio-friendly wrapper under ``opendht.aio``. +""" + +from ._core import * # re-export core Cython bindings + +__all__ = [name for name in dir() if not name.startswith("_")] diff --git a/python/opendht/aio.py b/python/opendht/aio.py new file mode 100644 index 00000000..9a78070e --- /dev/null +++ b/python/opendht/aio.py @@ -0,0 +1,336 @@ +""" +Asyncio-friendly wrapper for OpenDHT Python bindings. + +Usage: + + from opendht import aio as dht + + node = dht.DhtRunner() + await node.run() + await node.bootstrap("bootstrap.jami.net", "4222") + + key = dht.InfoHash.get("mykey") + await node.put(key, dht.Value(b"hello")) + vals = await node.get(key) + + async for v, expired in await node.listen(key): + ... +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Callable, Iterable, List, Optional, Tuple + +from . import _core as _dht + +# Re-export core types for convenience +from ._core import InfoHash, Value, ValueType, Where, SockAddr, PublicKey, PkId, DhtConfig, Identity + + +class _DoneOnce: + """Helper to prevent multiple resolution of a Future from repeated callbacks.""" + + __slots__ = ("_done",) + + def __init__(self) -> None: + self._done = False + + def mark(self) -> bool: + if self._done: + return False + self._done = True + return True + + +class _Listener: + """Async iterator around DHT listen callback. + + Yields tuples (Value, expired: bool). + """ + + __slots__ = ("_runner", "_token", "_loop", "_queue", "_closed") + + def __init__(self, runner: DhtRunner, key: InfoHash) -> None: + self._runner = runner + self._loop = runner._loop + self._queue: asyncio.Queue[Tuple[Value, bool]] = asyncio.Queue() + self._closed = False + + def _on_value(v: Value, expired: bool) -> bool: + if self._closed: + return False + self._loop.call_soon_threadsafe(self._queue.put_nowait, (v, expired)) + return True # continue listening + + self._token = self._runner._dht.listen(key, _on_value) + + def __enter__(self) -> _Listener: + return self + + def __exit__(self, exc_type, exc, tb) -> None: + self.cancel() + + def __aiter__(self) -> _Listener: + return self + + async def __anext__(self) -> Tuple[Value, bool]: + if self._closed: + raise StopAsyncIteration + item = await self._queue.get() + if item is None: + self._closed = True + raise StopAsyncIteration + return item + + def cancel(self) -> None: + if self._closed: + return + self._closed = True + try: + self._runner._dht.cancelListen(self._token) + finally: + # Unblock any pending __anext__ + self._queue.put_nowait(None) + + +class _GetStream: + """Async iterator around DHT get callback. + + Yields Value instances as they are found, then ends. + Raises on failure. + """ + __slots__ = ("_runner", "_loop", "_queue", "_closed", "_error") + + def __init__( + self, + runner: DhtRunner, + key: InfoHash, + where: Optional[Where], + filter: Optional[Callable[[Value], bool]], + ) -> None: + self._runner = runner + self._loop = runner._loop + self._queue: asyncio.Queue[Optional[Value]] = asyncio.Queue() + self._closed = False + self._error: Optional[BaseException] = None + + def _on_value(v: Value) -> bool: + # Called on DHT thread + if self._closed: + return False + self._loop.call_soon_threadsafe(self._queue.put_nowait, v) + return True + + def _on_done(ok: bool, _nodes: Iterable[Any]) -> None: + # Called on DHT thread + if self._closed: + return + if not ok: + self._error = RuntimeError("DHT get failed") + self._loop.call_soon_threadsafe(self._queue.put_nowait, None) + + self._runner._dht.get(key, _on_value, _on_done, filter, where) + + def __enter__(self) -> _GetStream: + return self + + def __exit__(self, exc_type, exc, tb) -> None: + self.cancel() + + def __aiter__(self) -> _GetStream: + return self + + async def __anext__(self) -> Value: + if self._closed: + raise StopAsyncIteration + item = await self._queue.get() + if item is None: + self._closed = True + if self._error is not None: + raise self._error + raise StopAsyncIteration + return item + + def cancel(self) -> None: + if self._closed: + return + self._closed = True + # Unblock any pending __anext__ + self._queue.put_nowait(None) + + +class DhtRunner: + """Async wrapper around opendht.DhtRunner. + + Methods are awaitable and won't block the event loop. + """ + + __slots__ = ("_dht", "_loop", "_config", "_bootstrap") + + def __init__( + self, + *, + loop: Optional[asyncio.AbstractEventLoop] = None, + core: Optional[_dht.DhtRunner] = None, + config: Optional[DhtConfig] = None, + bootstrap: Optional[Iterable[Tuple[str, Optional[str]]]] = None, + ) -> None: + try: + self._loop = loop or asyncio.get_running_loop() + except RuntimeError: + self._loop = asyncio.get_event_loop() + self._dht = core or _dht.DhtRunner() + self._config = config + # Keep a copy to avoid consuming a one-shot iterator + self._bootstrap = list(bootstrap) if bootstrap is not None else None + + async def run( + self, + id: Optional[Identity] = None, + is_bootstrap: bool = False, + port: int = 0, + ipv4: str = "", + ipv6: str = "", + config: Optional[DhtConfig] = None, + ) -> None: + cfg = config or self._config or DhtConfig() + # The core run spawns internal threads; call directly + self._dht.run(id, is_bootstrap, port, ipv4, ipv6, cfg) + # Optionally bootstrap after starting + if self._bootstrap: + for host, port_str in self._bootstrap: + self._dht.bootstrap(host, port_str) + + def bootstrap(self, host: str, port: Optional[str] = None) -> None: + # The core call is fast; call directly + self._dht.bootstrap(host, port) + + async def ping(self, addr: SockAddr) -> bool: + fut: asyncio.Future[bool] = self._loop.create_future() + done_once = _DoneOnce() + + def _done(ok: bool) -> None: + if done_once.mark(): + self._loop.call_soon_threadsafe(fut.set_result, bool(ok)) + + # Use non-blocking callback form and await completion. + self._dht.ping(addr, _done) + return await fut + + def get( + self, + key: InfoHash, + *, + where: Optional[Where] = None, + filter: Optional[Callable[[Value], bool]] = None, + ): + # Return a streaming iterator of values; it ends on completion and raises on failure. + return _GetStream(self, key, where, filter) + + async def getAll( + self, + key: InfoHash, + *, + where: Optional[Where] = None, + filter: Optional[Callable[[Value], bool]] = None, + ) -> List[Value]: + """Collect all values for a key and return them as a list.""" + results: List[Value] = [] + async for v in self.get(key, where=where, filter=filter): + results.append(v) + return results + + async def getFirst( + self, + key: InfoHash, + *, + where: Optional[Where] = None, + predicate: Optional[Callable[[Value], Any]] = None, + filter: Optional[Callable[[Value], bool]] = None, + ) -> Optional[Value]: + """Return the first value matching an (optionally async) predicate, or None.""" + with self.get(key, where=where, filter=filter) as stream: + async for value in stream: + if predicate is None: + return value + result = predicate(value) + if asyncio.iscoroutine(result): + result = await result + if result: + return value + return None + + async def put(self, key: InfoHash, val: Value, *, permanent: bool = False) -> bool: + fut: asyncio.Future[bool] = self._loop.create_future() + done_once = _DoneOnce() + + def _done(ok: bool, _nodes: Iterable[Any]) -> None: + if done_once.mark(): + self._loop.call_soon_threadsafe(fut.set_result, bool(ok)) + + self._dht.put(key, val, _done, permanent) + return await fut + + async def putSigned(self, key: InfoHash, val: Value, *, permanent: bool = False) -> bool: + fut: asyncio.Future[bool] = self._loop.create_future() + done_once = _DoneOnce() + + def _done(ok: bool, _nodes: Iterable[Any]) -> None: + if done_once.mark(): + self._loop.call_soon_threadsafe(fut.set_result, bool(ok)) + + self._dht.putSigned(key, val, _done, permanent) + return await fut + + async def putEncrypted( + self, + key: InfoHash, + to: InfoHash | PublicKey | PkId, + val: Value, + *, + permanent: bool = False, + ) -> bool: + fut: asyncio.Future[bool] = self._loop.create_future() + done_once = _DoneOnce() + + def _done(ok: bool, _nodes: Iterable[Any]) -> None: + if done_once.mark(): + self._loop.call_soon_threadsafe(fut.set_result, bool(ok)) + + self._dht.putEncrypted(key, val, to, _done, permanent) + return await fut + + def listen(self, key: InfoHash) -> _Listener: + return _Listener(self, key) + + async def cancelListen(self, listener: _Listener) -> None: + await listener.cancel() + + async def shutdown(self) -> None: + fut: asyncio.Future[None] = self._loop.create_future() + self._dht.shutdown(lambda: self._loop.call_soon_threadsafe(fut.set_result, None)) + await fut + + async def join(self) -> None: + await asyncio.to_thread(self._dht.join) + + async def __aenter__(self): + # Start the node on entering the async context (id/ports default to run() defaults) + if not self.isRunning(): + await self.run() + return self + + async def __aexit__(self, exc_type, exc, tb): + # Gracefully stop the node on context exit + if self.isRunning(): + try: + await self.shutdown() + finally: + await self.join() + return False # do not suppress exceptions + + def __getattr__(self, name: str) -> Any: + """Delegate unknown method calls to the wrapped DhtRunner instance.""" + return getattr(self._dht, name) + diff --git a/python/setup.py.in b/python/setup.py.in index 1e9dc733..f6133b9a 100644 --- a/python/setup.py.in +++ b/python/setup.py.in @@ -13,7 +13,7 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . -from setuptools import setup, Extension +from setuptools import setup, Extension, find_packages from setuptools.command.build_ext import build_ext as _build_ext import os @@ -63,8 +63,10 @@ setup(name="opendht", 'Programming Language :: Python :: 3.9', ], cmdclass=dict(build_ext=build_ext), - ext_modules = [Extension( - "opendht", + packages=find_packages(where='@CURRENT_SOURCE_DIR@'), + package_dir={"": "@CURRENT_SOURCE_DIR@"}, + ext_modules=[Extension( + "opendht._core", ["@CURRENT_SOURCE_DIR@/opendht.pyx"], language="c++", extra_compile_args=["-DMSGPACK_NO_BOOST", "-DMSGPACK_DISABLE_LEGACY_NIL", "-DMSGPACK_DISABLE_LEGACY_CONVERT", "-DOPENDHT_INDEXATION", "-Dopendht_EXPORTS"],