jsondb2.py

Version 0.1.0.1 · Source: modules/jsondb2.py

A JSON file wrapper with dictionary-style access, optional autosave, and lightweight async queue utilities.

Setup and storage model

Place jsondb2.py on the import path. JSON and queue operations use standard-library modules. Import also calls pmblue_update.self_update(); if the updater is absent, the bootstrap requires requests, downloads over the legacy HTTP URL, and writes pmblue_update.py in the working directory.

JsonFile(fp, **kwargs) reads the entire file into memory. A missing file is created with {}; its parent directory must already exist. Data is intended to be a JSON dictionary or list. Files are rewritten in full when saved, using compact JSON with ensure_ascii=False.

Explicit save example

from jsondb2 import JsonFile

settings = JsonFile("settings.json")
settings["theme"] = "dark"
settings["recent_files"] = ["report.csv"]
settings["recent_files"].append("summary.csv")
settings.save()

settings.reload()
print(settings["theme"])
print("theme" in settings)

Constructor options

Reading and writing

file.data exposes the loaded structure. Assigning data accepts only dictionaries/lists. file[key], assignment, deletion, membership, and iteration delegate to this structure. save() and reload() return None; reload() rereads disk without merging changes. Iteration uses the current memory state directly.

Autosave does not intercept every nested mutation. For example, file["items"].append(value) and nested deletion delegate directly to a list/dictionary and do not reliably trigger a write. Explicit save() after nested mutations is the predictable approach. Replacing the entire data value also does not recursively wrap the replacement immediately.

Current print(indent=2) only initializes its serialized value when auto_save=True; with defaults it raises an error. Use print(settings.data) or, when autosave is off, json.dumps(settings.data, indent=2).

Async queue utilities

Queue.add(coro) and Queue.add_sync(func, *args, **kwargs) return asyncio Tasks; await those tasks to observe completion and errors. await queue.add_and_wait(coro) runs a coroutine in the queue and returns its result. queue.qm() supplies a sync/async context manager that serializes entry within this process.

import asyncio
from jsondb2 import Queue

async def worker():
    await asyncio.sleep(0)
    return "finished"

async def main():
    queue = Queue()
    task = queue.add(worker())
    print(await task)

asyncio.run(main())

await queue waits for IDs that were registered when waiting started; await queue.wait_until_empty() waits while registered entries remain. Neither guarantees that newly scheduled tasks have entered the queue yet. JsonFile.save() discards the internally created Task when async_save=True, so it does not expose a reliable public flush/completion handle. Prefer the default synchronous save for code that must know persistence has completed.

AsyncProperty is a descriptor for async getters accessed with await obj.property; its wrapper also supports indexing before awaiting. Descriptor coroutine state is shared on the descriptor, so overlapping accesses are not a concurrency guarantee.

Persistence limits

Writes truncate and rewrite the live JSON file. There is no atomic replacement, transaction rollback, cross-process file lock, or merge between independent instances. Invalid JSON and file/serialization errors propagate to the caller. Use this wrapper for small local state with a single writer; the module updater's transactional installation does not change JsonFile's persistence behavior.

For relational data, see sqliteObj and aiosqliteObj.

Objects and method contracts

JsonFile is a file-backed container. Its fp identifies the file; auto_save, async_save, and reload_data store the constructor options. The data property exposes the actual in-memory dictionary or list, so changing it changes the wrapper's state. With autosave enabled, nested values can be internal container wrappers rather than plain dict/list objects.

Queue and QueueManager

Queue() tracks active entry IDs in q. active is a boolean property and len(queue) counts registered entries. The ID is registered when a scheduled task actually enters its QueueManager, so these values may still be empty immediately after queue.add(). QueueManager is normally obtained through queue.qm(); both with and async with serialize access, and waiting too long raises TimeoutError.

add(coro) takes an already-created coroutine; add_sync(func, *args, **kwargs) takes a callable plus its arguments. Both return Tasks, and awaited task results are the callable/coroutine's return value. Synchronous work is executed on the event-loop thread, so the queue does not make file writes nonblocking.

Worked example: autosave plus nested edits

This uses a temporary file and explicitly saves after append. That final save is deliberate: autosave reliably intercepts top-level assignments, but not all delegated list methods.

import os
import tempfile
from jsondb2 import JsonFile

with tempfile.TemporaryDirectory() as directory:
    path = os.path.join(directory, "preferences.json")
    preferences = JsonFile(path, auto_save=True)
    preferences["theme"] = "dark"
    preferences["recent"] = ["report.csv"]
    preferences["recent"].append("summary.csv")
    preferences.save()

    reread = JsonFile(path)
    print(reread.data)
    # {'theme': 'dark', 'recent': ['report.csv', 'summary.csv']}
    del preferences["theme"]
    reread.reload()
    print("theme" in reread)  # False

Worked example: await all queued work and errors

Keep the Task objects and await them with asyncio.gather. This observes completion even before those tasks have registered queue IDs, and propagates failures to the caller.

import asyncio
from jsondb2 import Queue

async def main():
    queue = Queue()
    events = []

    async def record(label):
        events.append("start " + label)
        await asyncio.sleep(0)
        events.append("end " + label)
        return label.upper()

    tasks = [queue.add(record(label)) for label in ("a", "b", "c")]
    results = await asyncio.gather(*tasks)
    print(results)  # ['A', 'B', 'C']
    print(events)  # Each task finishes its critical section before the next enters.

asyncio.run(main())

Custom serialization

A dump callback is useful with the default auto_save=False: pass dump=lambda data, stream: json.dump(data, stream, indent=2) for readable files. When autosave is enabled, a custom callback receives the internal nested wrappers and must handle them itself. Loading always uses json.load, so the callback must still produce valid JSON.