OpenDHT

OpenDHT is a lightweight Distributed Hash Table implementation providing an easy to use distributed in-memory data store.

Key Features

• Lightweight and scalable, designed for large networks and small devices
• High resilience to network disruption
• Public key cryptography layer providing optional data signature and encryption
• IPv4 and IPv6 support
• Python 3 bindings
• REST API with optional HTTP client+server with push notification support

Every node in the network can read and write values to the store. Values are distributed over the network with redundancy.

Installation

Install on Linux, macOS and Windows with:

pip install opendht

Quick Example

Using the simple, blocking API:

import opendht as dht

node = dht.DhtRunner()
node.run()

# Join the DHT network through any running node,
# here using a known bootstrap node.
node.bootstrap("bootstrap.jami.net", "4222")

# blocking call (provide callback arguments to make the call non-blocking)
node.put(dht.InfoHash.get("unique_key"), dht.Value(b'some binary data'))

results = node.get(dht.InfoHash.get("unique_key"))
for value in results:
    print(value)

Or using asyncio:

import asyncio
import opendht.aio as dht

async def dht_async_demo(key_str: str):
    # Start a new node using an async context manager.
    # It is also possible to call run()/await shutdown() manually.
    async with dht.DhtRunner(
        bootstrap=(("bootstrap.jami.net", "4222"),)
    ) as node:
        # compute key hash
        key = dht.InfoHash.get(key_str)

        # put data, waiting for completion
        await node.put(key, dht.Value(b'tata data'))

        # get all values at key
        results = await node.getAll(key)
        for value in results:
            print(value)
        
        # same operation, but stream values as they come from the network
        with node.get(key) as results:
            async for value in results:
                print(value)

        # listen for change of values at key
        with node.listen(key) as values:
            async for value, expired in values:
                print(value)
                if value.data == b'tata data':
                    break

asyncio.run(dht_async_demo("unique_key"))

Node State Persistence

Set the state file path before starting a node:

import threading
import opendht as dht

config = dht.DhtConfig()
config.setPersistPath("dht-state.msgpack")
node = dht.DhtRunner()
node.run(config=config)
stopped = threading.Event()
node.shutdown(stopped.set)
stopped.wait()
node.join()

The node loads its saved DHT node ID, known peers, and stored values on startup, and saves state during shutdown(). Wait for the shutdown callback before calling join() or reusing the state path; join() alone does not save state. Port hints are stored in a sibling file named <path>_port.txt; saved ports are reused when available. The parent directory must already exist and be writable. Use a different state path for each node.

Persistence is disabled by default. config.setPersistPath("") disables it explicitly. This state does not include private keys or certificates; persist cryptographic identities separately as described below.

Identity Persistence

Keys can be imported from PEM or DER bytes. Private keys export PKCS#8 PEM; public keys export DER. Certificates export their complete PEM chain.

import opendht as dht

identity = dht.Identity.generate("my-node")

private_key_pem = identity.key.serialize(password)
certificate_pem = bytes(identity.certificate)
public_key_der = bytes(identity.publickey)

restored = dht.Identity(
    dht.PrivateKey(private_key_pem, password),
    dht.Certificate(certificate_pem),
)
public_key = dht.PublicKey(public_key_der)
node = dht.DhtRunner()
node.run(id=restored)

Here password is an application-supplied string. Save the exported bytes using your application's storage, protect private key files, and keep the password separate from the key. bytes(private_key) and private_key.serialize() without a password export unencrypted private keys. Empty or malformed key data, incorrect passwords, and passwords containing NUL characters raise exceptions. Exporting an uninitialized key raises ValueError.

Value Metadata

The following properties are readable and writable on Value, including values returned by get() and listen():

value = dht.Value(b"some binary data")
value.type = 42
value.seq = 1
value.priority = 1
value.pushType = "audioCall"

Set metadata before publishing the value. Numeric values outside the native integer range raise OverflowError.

value.isSigned() reports whether a public key and signature are present; it does not verify the signature. value.isEncrypted() reports whether the value currently contains ciphertext. Values delivered by get() and listen() are transparently decrypted when possible, so decrypted values return False from isEncrypted(), even if they were published using putEncrypted(). An uninitialized PublicKey() cannot be used as an encryption recipient and raises ValueError.

Register custom ValueType instances after run() and before join(). registerType() raises RuntimeError when the runner is stopped, TypeError for None, and ValueError for an uninitialized type.

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.

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, type registration, logging controls, run(), blocking operations, or join() from them. The asyncio wrapper schedules status callbacks on its event-loop thread, where snapshot getters are safe to use.

Call node.connectivityChanged() after a network-interface or IP-address change. It queues native connectivity recovery, including recontacting peers and renewing subscriptions. It is also available on the asyncio runner as a synchronous, nonblocking method.

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.

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.

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.

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

License

Copyright (c) 2014-2026 Savoir-faire Linux Inc.

OpenDHT is released under the MIT License. See LICENSE for details.