Reference · development API

The Python API

Database is the Rust-native engine for immutable document logs and rebuildable Wasm views. View identifies the complete compiled computation. There is one storage and computation path.

View and packaged assets

View(module, *, sha256=None)
View.bundled(name="metrics") → View
portable_assets() → importlib.resources Traversable

Import these names from foldlitedb. module accepts Wasm bytes or a filesystem path. A View stores immutable module bytes and their sha256; an optional digest must match. Modules range from 1 byte through 16 MiB. The native database validates exports, imports, ABI, and descriptor when opening.

bundled accepts "metrics" or "catalogue" and verifies the manifest digest. Metrics returns {count,sum,min,max,mean}; catalogue returns {count,items}. See accepted payloads. portable_assets() exposes modules, manifests, and browser hosts; copy files with each resource's read_bytes() method.

PortableView is an exact alias of View. It does not select another engine.

Document: a named source observation

Document(id, payload, position=None, source="default")

Import Document from foldlitedb to construct records for ingest or ingest_many. Construction alone writes nothing. id is a stable, nonempty string naming one observation within its source; payload is its finite JSON value. The same ID in another source names a different observation.

Supply position when the upstream source already has ordered offsets. New positions must advance that source, with gaps allowed. If the source has stable IDs but no offsets, omit position: ingestion assigns one automatically. An identical named retry with position=None reuses the original position, even after newer records arrive. Input objects are not mutated; use documents() to export their assigned positions. Records returned by add, add_many, or documents always have resolved integer positions.

The fields are id, payload, position, and source. Returned payloads are independent copies; changing a copy cannot update a stored document. Reusing an ID or position with conflicting content raises DocumentConflict. Append a new named observation to express a change.

Database

Database(path, view=None)

A Rust-native database executing one complete Wasm view. view accepts a View, Wasm bytes, or a module path; omit it for bundled metrics. path accepts a filesystem path or ":memory:". Parent directories must already exist. A persistent path is an append-only document log, with .cache and .lock siblings. Exactly one active owner may open it.

Use a context manager or close(). Clean close checkpoints persistent derived state and releases ownership. Closing is idempotent; later operations raise DatabaseError. PortableDatabase is an exact alias.

Reopen validates the complete log and reuses a compatible cache, then replays any uncached suffix. A missing or invalid cache rebuilds. Different module bytes automatically rebuild the view from the same documents. Reconstruction creates a new cursor generation; a valid checkpoint preserves generation across reopen. Old external-database files are not this log format.

from foldlitedb import Database, View

with Database(":memory:", View.bundled("metrics")) as db:
    original = db.add({"metric": "loss", "value": 4}, id="a")
    assert db.add(original.payload, id="a") == original
    snapshot = db.query()
    db.add({"metric": "loss", "value": 8})
    assert db.changes(since=snapshot.cursor).rows[0].result["mean"] == 6
    retained = tuple(db.documents())
    assert db.stats()["documents"] == 2

with Database(":memory:") as replay:
    assert replay.ingest(retained).inserted == 2
    assert replay.ingest(retained).replayed == 2

Append and replay

db.add(payload, *, id=None, source="default", position=None) → Document
db.add_many(payloads, *, source="default") → tuple[Document, ...]
db.ingest(documents) → IngestResult
db.ingest_many(documents, *, batch_size=100) → iterator[IngestResult]

add appends one finite JSON payload and returns its resolved Document(id, payload, position, source). Omitted IDs are generated; omitted positions start at zero and advance per source. An identical named add or replay at an existing source position reuses the original document. Conflicting reuse fails; previously unseen positions must advance that source. IDs and sources are nonempty strings of at most 4096 UTF-8 bytes; positions range from zero through 2**53 - 1.

add_many eagerly consumes one iterable of payloads and commits atomically. Its returned records preserve input order. Every member is one payload, including an entire array; empty input returns (). Retrying unnamed payloads appends new observations. Documents are immutable; append a new observation to express a change.

ingest takes Document instances with stable identities, payloads, and optional upstream positions. Its result contains inserted, replayed, and sequence. Identical replay performs no map work.

ingest_many is lazy: you must iterate its results to write anything. Each yielded result commits a separate batch of 1–1024 records, defaulting to 100. A later error preserves earlier committed batches; the complete import is not one atomic transaction.

An atomic batch has at most 1024 documents and 4096 emissions. The complete map request, including its envelope and newly inserted payloads, must fit 256 KiB; per-group add requests, states, and results must fit their own buffers. The log frame limit is 8 MiB. Reduce batch size for larger payloads. Validation, map, reducer, and limit errors leave the current batch uncommitted; successful persistent batches are synced before acknowledgement.

Finite JSON uses IEEE-754 doubles and at most 64 nesting levels. Python rejects integer inputs not exactly representable as doubles. Nonfinite values and invalid Unicode scalar strings fail. Use strings for arbitrary-precision identifiers.

Retry-safe source batches

If your telemetry export has offsets, retain them along with the source and stable reading IDs. Fully iterating the import below commits 25 batches; resending an old batch counts no readings twice:

from foldlitedb import Database, Document

source_records = [
    {"id": f"gateway-07:{offset}", "offset": offset,
     "metric": "temperature.c", "value": 20 + (offset % 20) * 0.25}
    for offset in range(2500)
]
records = [Document(r["id"], {"metric": r["metric"], "value": r["value"]},
                    position=r["offset"], source="gateway-07")
           for r in source_records]
with Database(":memory:") as imported:
    results = list(imported.ingest_many(records, batch_size=100))
    assert sum(result.inserted for result in results) == 2500
    retry = imported.ingest(records[100:200])
    assert retry.inserted == 0 and retry.replayed == 100
    assert list(imported.rows())[0].result["count"] == 2500

If upstream provides IDs but no offsets, use named records with automatic positions instead. Do not invent new offsets from each export's row numbers:

named = [
    Document("probe-a:reading-1", {"metric": "temperature.c", "value": 20}, source="probe-a"),
    Document("probe-a:reading-2", {"metric": "temperature.c", "value": 21}, source="probe-a"),
]
with Database(":memory:") as automatic:
    assert sum(r.inserted for r in automatic.ingest_many(named)) == 2
    automatic.add({"metric": "temperature.c", "value": 22}, id="probe-a:reading-3", source="probe-a")
    assert automatic.ingest(named).replayed == 2
    assert [record.position for record in automatic.documents()] == [0, 1, 2]

For JSONL, construct each Document while iterating the file and pass that iterator to ingest_many. Keep source order, and retain resolved records when exporting. The complete telemetry example verifies summaries against a Python oracle, follows live changes, and reopens after deleting the cache.

Inspect a payload's map output

db.preview(payload) → list[{"key": JSON, "value": JSON}]

preview runs only the current view's map function without appending a document or changing source progress, rows, or history. It does not run the reducer or guarantee that a later ingestion batch fits its limits. Use it to inspect unexpected zero-emission payloads:

with Database(":memory:") as inspection:
    assert inspection.preview({"metric": "temperature.c", "value": 21}) == [
        {"key": "temperature.c", "value": 21},
    ]
    assert inspection.preview({"kind": "heartbeat"}) == []
    assert inspection.stats()["documents"] == 0

Zero emissions are intentional filtering, not a rejection: ingestion still retains a valid document and advances its source progress. The bundled metrics view ignores shapes without a string metric and finite numeric value. Validate your application schema before ingestion if every retained document should contribute a reading.

Rows, snapshots, and changes

db.rows(*, key=<omitted>, limit=100) → iterator[Row]
db.query(*, key=<omitted>, limit=100, cursor=None, after=None) → Snapshot
db.changes(*, since, limit=100) → ChangePage

rows lazily follows snapshot pages, pinning the committed prefix on first iteration. limit is a page size from 1 through 1024, not a total row limit. Omit key for every group or supply any JSON key, including null, for exact filtering. Rows follow canonical JSON lexical order consistent with the browser. No operation lock spans a yield into application code.

A Row has key, reducer state, finalized result, and committed sequence. A Snapshot has rows, cursor, after, and complete. Continue with both returned cursor and after token, preserving the key filter. New ingestion does not change the pinned snapshot.

with Database(":memory:") as db:
    db.add_many([
        {"metric": "a", "value": 2},
        {"metric": "b", "value": 8},
    ])
    page = db.query(limit=1)
    snapshot_cursor = page.cursor
    rows = list(page.rows)
    while not page.complete:
        page = db.query(limit=1, cursor=page.cursor, after=page.after)
        rows.extend(page.rows)
    assert [row.key for row in rows] == ["a", "b"]

    db.add({"metric": "a", "value": 4})
    changes = db.changes(since=snapshot_cursor, limit=1)
    assert changes.rows[0].result["mean"] == 3
    while not changes.complete:
        changes = db.changes(since=changes.cursor, limit=1)

changes returns committed group postimages after a snapshot or change cursor. since=0 reads retained history from the start. A page has rows, continuation cursor, and complete; continue with that cursor. Repeated changes to a key in one batch coalesce. A page can split group updates from one batch. Zero-emission documents still advance progress.

Cursors bind to the database generation and exact module. Checkpointed reopen can reuse them; rebuild, cache reconstruction, or module replacement expires them. A new database or browser worker cannot reuse another's cursor. There is no history compaction API.

A persistent changes consumer that recovers

Save a consumer's cursor together with the application state it describes. Change rows contain whole resulting group states, so replace each key's result. If the cursor expires, collect a fresh stable snapshot and replace the entire application state. Replaying from since=0 onto existing totals would double count.

from pathlib import Path
from tempfile import TemporaryDirectory
from foldlitedb import CursorExpired

def collect_snapshot(db):
    page = db.query(limit=1)
    state, cursor = {}, page.cursor
    while True:
        state.update((row.key, row.result) for row in page.rows)
        if page.complete:
            return state, cursor
        page = db.query(limit=1, cursor=page.cursor, after=page.after)

def poll_changes(db, state, cursor):
    try:
        while True:
            page = db.changes(since=cursor, limit=1)
            state.update((row.key, row.result) for row in page.rows)
            cursor = page.cursor
            if page.complete:
                return state, cursor
    except CursorExpired:
        return collect_snapshot(db)  # New dict replaces all old application state.

with TemporaryDirectory() as directory:
    path = Path(directory) / "consumer.fold"
    with Database(path) as db:
        db.add_many([{"metric": "cpu.percent", "value": 20},
                     {"metric": "temperature.c", "value": 21}])
        state, cursor = collect_snapshot(db)
        db.add({"metric": "temperature.c", "value": 23})
        state, cursor = poll_changes(db, state, cursor)
        assert state["temperature.c"]["mean"] == 22
    with Database(path) as db:  # Clean close kept this cursor valid.
        assert db.changes(since=cursor).rows == ()
    Path(str(path) + ".cache").unlink()
    with Database(path) as db:
        previous_state = state
        state, cursor = poll_changes(db, state, cursor)
        assert state is not previous_state
        assert state == {row.key: row.result for row in db.rows()}

The runnable telemetry collector also persists its consumer checkpoint and proves that cache reconstruction preserves every resolved source record and summary.

Checkpoint, rebuild, export, and inspect

db.checkpoint() → dict
db.rebuild() → dict
db.documents(*, limit=100) → iterator[Document]
db.stats() → dict
db.close()

checkpoint atomically persists the current derived snapshot and returns stats plus checkpointed: true. Clean close also checkpoints. The cache is optional and is not rewritten after every append. Encoded snapshots are capped at 1 GiB. Checkpoint failures are reported, including during close; the synced source log remains intact and close releases ownership even on failure. rebuild replays original log batches through the current module, starts a new cursor generation, writes a checkpoint, and returns stats plus rebuilt: true. Documents remain unchanged.

with Database("maintenance.fold") as db:
    db.add_many([{"metric": "loss", "value": 4}, {"metric": "loss", "value": 8}])
    expected = [(row.key, row.state, row.result) for row in db.rows()]
    assert db.checkpoint()["checkpointed"] is True
    assert db.rebuild()["rebuilt"] is True
    assert [(row.key, row.state, row.result) for row in db.rows()] == expected

documents exports resolved records in original ingestion order at a snapshot pinned on first iteration. Page size is 1–1024. Back up these records or the log and retain exact view artifacts; the cache can be regenerated. Re-import through ingest or ingest_many.

stats includes documents, groups, sequence, cursor, generation, module_sha256, descriptor, and source checkpoints. storage is "append-only-log" or "memory". The cache object reports status, bytes, and log_offset; replay reports processed batches and documents; log reports bytes, batches, and recovered_tail_bytes.

Engine sequence counts inserted documents and changed group states, rather than documents alone. Reopening always verifies the complete log, even with a valid cache. Current retained documents, indexes, and view history grow in host memory; there is no compaction or bounded total-memory guarantee.

cache.status describes startup or the most recent checkpoint operation, not whether the in-memory view is usable. For example, status="missing", bytes=0, and replay.documents=2513 is healthy after rebuilding from a missing cache. The status stays "missing" until checkpoint() or clean close writes a cache. cache.bytes is the cache file size inspected or last written, not the memory used by the view; cache.log_offset is the reused or checkpointed log prefix.

Errors and recovery

DatabaseError extends ValueError. Validation and Wasm failures raise it; conflicting replay, source ordering, and invalid cursors use DocumentConflict, SourceOrderError, and CursorExpired where identified. A missing compiled extension raises an actionable RuntimeError.

Opening an already-owned log fails. Complete log corruption fails rather than skipping records; only an incomplete final frame is recoverable as an interrupted append. A cache failure can be repaired by reconstruction. If an I/O error makes an append's acknowledgement uncertain, close, reopen, and retry using explicit resolved identities. Keep backups of source data.