nedb-engine-client 2.8.2

Async Python client for nedbd — the nedb-engine server daemon

nedb-engine-client

Async Python client for nedbd — the NEDB server daemon.

Connect to any running nedbd instance — local or remote — with a clean async API. No engine code embedded, no Rust toolchain required. Just HTTP.

---

Install

pip install nedb-engine-client

Requires Python ≥ 3.8 and httpx.

---

Quick start

import asyncio
from nedb_client import NedbClient

async def main():
    async with NedbClient("http://127.0.0.1:7070", db="mydb") as db:

        # Write a document
        await db.put("blocks", "618000", {
            "height": 618000,
            "hash": "000000000000000000024bead8df69990852c202db0e0097c1a12ea637d7e96d",
            "tx_count": 2734,
        })

        # Query with NQL
        rows = await db.query("FROM blocks ORDER BY height DESC LIMIT 10")

        # Time-travel: what did the DB look like at seq 100?
        old = await db.query("FROM blocks AS OF 100 WHERE height > 600000")

        # Bi-temporal: what was true on 2024-06-15?
        valid = await db.query('FROM policy VALID AS OF "2024-06-15"')

        # Causal trace: why was this written?
        trace = await db.query("FROM blocks TRACE caused_by")

        # BLAKE2b Merkle head — changes on every write, anchorable
        head = await db.head()
        print(f"head: {head}")

        # Tamper-evidence check across all objects
        report = await db.verify()
        assert report["ok"], "tamper detected!"

asyncio.run(main())

---

nedbd — start the server

pip install nedb-engine

# v1 AOF engine (default)
nedbd --data ./data

# v2 DAG engine (recommended — instant cold start, tamper-evident)
NEDBD_DAG=1 nedbd --data ./data

# With AES-256-GCM encryption
NEDBD_DAG=1 NEDB_TMK=<32-byte-hex> nedbd --data ./data

# With natural-language planning (see `cast` below)
#   needs a build with --features cast, plus model.cast in the data dir
NEDBD_DAG=1 NEDBD_CAST=1 nedbd --data ./data

# Check health
curl http://127.0.0.1:7070/health
# {"ok":true,"version":"2.8.0","service":"nedbd","encrypted":true}

---

API reference

Client lifecycle

# Async context manager (recommended)
async with NedbClient(url, db=name, token=token) as db:
    ...

# Manual
db = NedbClient(url="http://127.0.0.1:7070", db="mydb", token="secret")
await db.open()
await db.close()

Writes

Method	Description
await db.put(coll, id, doc, **opts)	Write a document
await db.delete(coll, id)	Tombstone delete (history preserved in DAG)
await db.batch(ops)	Batch put/del in one HTTP round-trip
await db.create_index(coll, field)	Create sorted index for ORDER BY

Put options:

await db.put("claims", "c1", {"fact": "..."}, 
    caused_by=["abc123hash"],   # DAG causal provenance
    valid_from="2024-01-01",    # bi-temporal valid window
    valid_to="2024-12-31",
    evidence="sensor-42",       # human-readable provenance note
    confidence=0.95,            # confidence score 0–1
    idem="dedup-key",           # idempotency key
)

Reads

Method	Description
await db.get(coll, id)	Fetch current version of a document
await db.query(nql)	NQL query → list of dicts
await db.query_full(nql)	NQL query → full response (rows + seq + head)

NQL — NEDB Query Language

FROM <collection>
  [AS OF <seq>]                  transaction time (when was it written?)
  [VALID AS OF "<date>"]         valid time (when was it true in the world?)
  [WHERE field = value [AND ...]] op: = != < <= > >=
  [ORDER BY field [DESC]]
  [LIMIT n]
  [GROUP BY field COUNT|SUM|AVG|MIN|MAX]
  [TRACE caused_by [REVERSE]]    causal graph traversal
  [SEARCH "text"]                full-text search

Cast — natural language into NQL

Method	Description
await db.cast(prompt, execute=False)	English → dict (the plan)
await db.cast_nql(prompt)	English → just the NQL string

Requires a daemon built with --features cast and started with --cast. A 3.33M-parameter model (nedb-cast-slm) runs server-side on CPU — no API key, no per-token bill.

plan = await db.cast("orders over 100")
# {'prompt': 'orders over 100',
#  'nql': 'FROM orders WHERE total > 100',
#  'valid': True,
#  'collection': 'orders',
#  'collection_known': True,
#  'collections': ['orders'],
#  'executed': False,
#  'seq': 3, 'head': '262fd9…'}

execute defaults to False on purpose. You get a plan to look at, not a query that already ran:

if plan["valid"] and plan["collection_known"]:
    rows = await db.query(plan["nql"])          # review, then run

run = await db.cast("orders over 100", execute=True)
run["count"]   # 2
run["rows"]    # [...]

That default matters. A real miss from a real run:

prompt   "paid orders over 100"
nql      FROM orders WHERE status = "paid" LIMIT 100      ← wrong
correct  FROM orders WHERE status = "paid" AND total > 100

It read "over 100" as LIMIT 100. The count still came back correct by coincidence — both paid orders exceeded 100. Multi-predicate WHERE is the model's weakest clause (85.1% eval / 61.2% holdout). Show the plan to a human when you can.

Errors are explicit, never empty results. The server checks the plan against the collections this database actually has:

try:
    await db.cast("show me all stylists")
except NedbError as e:
    e.status    # 422
    e.message   # collection "stylists" does not exist in "shop" (generated: 'FROM stylists')

Zero rows would read as "no matching data" — a lie, when the truth is the collection was imagined. Status 501 means the daemon was built without the feature, so you can detect the capability rather than guess at it.

Integrity

Method	Description
await db.verify()	BLAKE2b tamper-evidence check across all objects
await db.head()	Current Merkle root — changes on every write
await db.seq()	Current global sequence number
await db.log(limit)	Recent write log
await db.checkpoint()	Explicit checkpoint (no-op on v2 DAG)

Server management

Method	Description
await db.health()	Server health — version, databases, encryption
await db.ping()	Boolean reachability check
await db.list_databases()	All databases on this server
await db.create_database()	Create this database explicitly
await db.drop_database()	Drop this database (irreversible)

---

Error handling

from nedb_client import NedbClient, NedbError

async with NedbClient("http://127.0.0.1:7070", db="mydb") as db:
    try:
        await db.put("coll", "id", {"data": "value"})
    except NedbError as e:
        print(f"HTTP {e.status}: {e.message}")

Queries on missing collections return [] rather than raising — resilient by default.

---

Links

• Engine: pip install nedb-engine · npm install nedb-engine
• JS/TS client: npm install nedb-engine-client
• Cast model: nedb-cast-slm — the 3.3M-param planner behind db.cast()
• Source: github.com/Eth-Interchained/nedb
• Studio: studio.interchained.org

---

© INTERCHAINED, LLC · GPL-3.0-or-later · Built with Hyperagent