jsonjsdb

Python library for JSONJS databases with full CRUD support and relational queries.

Features

• Read & Write: Full CRUD operations
• Typed API: Optional TypedDict support with autocompletion
• Relational queries: having.{table}(id) for one-to-many and many-to-many
• Filtering: where() with operators (==, !=, >, in, is_null, etc.)
• TypeScript compatible: Same file format as the TypeScript jsonjsdb library

Installation

pip install jsonjsdb

Quick Start

from jsonjsdb import Jsonjsdb

db = Jsonjsdb("path/to/db")

# Read
user = db["user"].get("user_1")
active = db["user"].where("status", "==", "active")

# Write
db["user"].add({"id": "u1", "name": "Alice", "tag_ids": []})
db["user"].update("u1", name="Alice Updated")
db.save()

Typed Access

With TypedDict (dict-style)

from typing import TypedDict
from jsonjsdb import Jsonjsdb, Table

class User(TypedDict):
    id: str
    name: str
    tag_ids: list[str]

class MyDB(Jsonjsdb):
    user: Table[User]

db = MyDB("path/to/db")
user = db.user.get("user_1")  # Returns User | None
print(user["name"])           # Dict-style access

With Dataclass (attribute-style)

from dataclasses import dataclass
from jsonjsdb import Jsonjsdb, Table

@dataclass
class User:
    id: str
    name: str
    tag_ids: list[str]

# Pass entity_type to get dataclass instances
table: Table[User] = Table("user", entity_type=User)

user = table.get("user_1")    # Returns User dataclass
print(user.name)              # Attribute-style access

table.add(User(id="u2", name="Bob", tag_ids=[]))

API Reference

CRUD

db.user.add({"id": "u1", "name": "Alice", ...})  # Add row (id required)
db.user.add_all([...])                           # Add multiple rows (batch)
db.user.upsert({"id": "u1", ...})                # Add or update → bool (True=added)
db.user.upsert_all([...])                        # Insert-or-replace multiple rows (single rebuild)

db.user.get("u1")                                # → User | None
db.user.get_many(["u1", "u2"])                   # → list[User] (only requested rows)
db.user.get_by("email", "alice@test.com")        # → User | None (by column)
db.user.exists("u1")                             # → bool
db.user.all()                                    # → list[User]
db.user.count                                    # → int (number of rows)
db.user.is_empty                                 # → bool

db.user.update("u1", name="New Name")            # Update fields
db.user.update_many(["u1", "u2"], status="x")   # Batch update → int (count)
db.user.remove("u1")                             # → bool
db.user.remove_all(["u1", "u2"])                 # → int (count)
db.user.remove_where("status", "==", "inactive") # → int (count)

Filtering

db.user.where("status", "==", "active")          # Equality
db.user.where("age", ">", 18)                    # Comparison (>, >=, <, <=)
db.user.where("status", "in", ["a", "b"])        # In list
db.user.where("email", "is_null")                # Null check (is_not_null)

db.user.ids_where("status", "==", "active")      # → list[str] (IDs only, faster)

Relations

db.email.having.user("user_1")      # One-to-many: where user_id == "user_1"
db.user.having.tag("tag_1")         # Many-to-many: where tag_ids contains "tag_1"
db.folder.having.parent("folder_1") # Hierarchy: where parent_id == "folder_1"

db.email.ids_having.user("user_1")  # Same as above, returns IDs only (faster)

Save / New Database

db.save()                # Save to original path
db.save("new/path")      # Save to new location

db = MyDB()              # Create empty in-memory DB
db.user.add({...})
db.save("path/to/db")    # Path required on first save

Evolution Tracking

Changes are automatically tracked when saving. An evolution.json file logs all additions, deletions, and updates:

# Tracking enabled by default
db.save()

# Disable tracking
db.save(track_evolution=False)

# Skip .json.js files (faster, smaller output)
db.save(write_js=False)

# Use Excel as source (for easy editing of logs)
db.save(evolution_xlsx=Path("path/to/evolution.xlsx"))

# Override timestamp for deterministic outputs (useful for testing)
db.save(timestamp=1741186800)

Cascade Filtering

When a parent entity is added or deleted, all child entities are also added/deleted. By default, this creates noise in the evolution log. Use parent_relations to automatically filter out cascade entries:

db.save(
    parent_relations={
        "variable": "dataset",    # variable.dataset_id → dataset
        "freq": "variable",       # freq.variable_id → variable
    }
)

With cascade filtering:

• Adding a dataset with 50 variables logs only 1 entry (the dataset add)
• Deleting a dataset logs only the parent delete, not all child deletes
• Updates are always logged (no filtering)
• Explicit child additions (to existing parent) are still logged

When evolution_xlsx is provided:

• The xlsx file becomes the source of truth (read from xlsx if it exists)
• User edits made in Excel are preserved on subsequent saves
• Both evolution.json and evolution.xlsx are written to stay in sync

Evolution format:

[
  {
    "timestamp": 1741186800,
    "type": "add",
    "entity": "user",
    "entity_id": "user_2",
    "parent_entity_id": null,
    "variable": null,
    "old_value": null,
    "new_value": null,
    "name": null
  },
  {
    "timestamp": 1741186800,
    "type": "update",
    "entity": "variable",
    "entity_id": "var_1",
    "parent_entity_id": "ds_1",
    "variable": "name",
    "old_value": "Old Name",
    "new_value": "New Name",
    "name": null
  }
]

Runtime Fields

Exclude fields from persistence (in-memory only):

from jsonjsdb import Table

# Option 1: Via constructor
table: Table[dict] = Table("user", runtime_fields={"_seen", "_processed"})

# Option 2: Via subclass
class UserTable(Table[User]):
    runtime_fields = {"_seen", "_processed"}

table.add({"id": "1", "name": "Alice", "_seen": True})

table.get("1")["_seen"]           # → True (in memory)
table.get_persistable_df()        # → DataFrame without _seen
# On save(), runtime_fields are automatically excluded

DataFrame Access

Inserts are buffered and materialized on the next read, so reach the underlying Polars DataFrame through table.df rather than any internal attribute. Reading it flushes pending rows; assigning to it replaces the whole frame, rebuilds the id index and applies the entity's storage schema.

table.df                                        # → DataFrame, buffered rows included
table.df = table.df.with_columns(...)           # add computed columns
table.df = table.df.join(other, on="id")        # join metadata in
table.df = table.df.filter(pl.col("keep"))      # drop rows

Rows still buffered when a frame is assigned are replaced by it, as they would be by any whole-frame overwrite. A frame the storage schema rejects raises and leaves the table untouched.

File Format

• __table__.json — Index of tables with metadata
• {table}.json — Data as array of objects
• {table}.json.js — Same data for browser (JavaScript)
• evolution.json — Change history (auto-generated on save)

Column Conventions

License

MIT