Installation

omni-json-db has zero required dependencies. It runs on a clean Python install using built-in pure-Python fallbacks, so this is all you need:

pip install omni-json-db

For the fastest, full-featured build — C-accelerated serializers(orjson / msgpack), extra compression codecs(zstandard / lz4 / brotli), YAML value formats, and the BTrees key-index backend — install the full extra:

pip install "omni-json-db[full]"

Or pick only what you need:

pip install "omni-json-db[speed]"        # faster serializers only (drop-in, no new features)
pip install "omni-json-db[compression]"  # zstd / lz4 / brotli value compression
pip install "omni-json-db[yaml]"         # YAML
pip install "omni-json-db[btree]"        # B-tree key-index backend

Basic

from omni_json_db import JDb
# Initialize the database from file
# Key-Value is Json+MsgPack without compression
jdb = JDb("example.jdb")

# Store data
jdb["user1"] = {"name" : "Ryan", "role": "Developer"}

# Retrieve data
user = jdb["user1"]
print(user["name"], user["role"]) # Output: Ryan Developer

All standard dict methods work: keys(), values(), items(), get(), pop(), setdefault(), update().

In-Memory Mode

from omni_json_db import JDb
# Initialize the database in memory
# Key-Value is Json+MsgPack without compression
jdb1 = JDb()

# Store data
jdb1 += {"user1" : {"name" : "Joe", "role": "Senior Developer"}}

# Retrieve data
print(jdb1["user1"]["name"]) # Output: Joe

# create 2nd JDb sharing same memory
jdb2 = JDb(jdb1)

# Store data to 2nd JDb
jdb2["user2"] = {"name" : "Kathy", "role": "CEO"}

# newly inserted data (by 2nd JDb)
print(jdb1["user2"]["name"]) # Output: Kathy

Query Engine

from omni_json_db import JDb
# Initialize the database in memory
# Key-Value is Json+Marshal with no compression
jdb = JDb(data_type="J+M")

# insert many records without key
jdb += [{'name': 'John', 'age': 22}, {'name': 'John', 'age': 37}, \
         {'name': 'Bob', 'age': 42}, {'name': 'Megan', 'age': 27}]

# get all records from database
print(jdb[:]) # print all records from jdb

# show table
jdb.show()

# Use FUNCTION to find record(s) matching the name 'John'
matches = jdb.find(FUNC=lambda key,val: val['name'] == 'John')
print(matches) # Output : {'0': {'name': 'John', 'age': 22}, '1': {'name': 'John', 'age': 37}}

# Use Regex to find record(s) matching the name 'John' or 'Bob'
matches = jdb.find(RE='John|Bob')
print(matches) # {'0': {'name': 'John', 'age': 22}, '1': {'name': 'John', 'age': 37}, '2': {'name': 'Bob', 'age': 42}}

Condition operators: EQ, NE, GT, LT, GTE, LTE, HAS, RE, RE2, FUNC, AND, OR, NOR, NOT, NAND, SIZE, ANY, ALL, NONE, IHAS, NHAS, EXISTS, TYPE, MOD, BETWEEN, NEAR, MATCH, SW, EW, NIN, ANYIN.

Transform operators: ABS, CEIL, FLOOR, ROUND, FLOAT, INT, NEG, STR, AVG, STD, MAX, MID, MIN, SUM, FIRST, LAST, LEN, SORT, UNIQUE, LOWER, UPPER, STRIP.

See More Query Examples or Pythonic Query Examples

Dataclass Objects

Any @dataclass with an id field can be written and read directly. The object is flattened into a plain dict — id is lifted out to become the record key — so the stored record is an ordinary one: still queryable, still exportable to CSV, still readable under every data_type.

from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from enum import Enum
from typing import List, Optional
from omni_json_db import JDb, Query

class Role(str, Enum):
    ADMIN = 'admin'
    USER  = 'user'

@dataclass
class Address:
    city: str
    zipcode: str = ''

@dataclass
class User:
    id: str                                       # required: becomes the record key
    name: str
    age: int = 0
    role: Role = Role.USER
    tags: List[str] = field(default_factory=list)
    address: Optional[Address] = None
    joined: Optional[date] = None
    balance: Decimal = Decimal('0')

jdb = JDb('users.jdb')

# --- Create ---
jdb += User('u1', 'Alice', 30, Role.ADMIN, ['python'], Address('Taipei', '110'), date(2024, 5, 1), Decimal('12.50'))
jdb += [User('u2', 'Bob', 25), User('u3', 'Charlie', 35)] # batch: one transaction
jdb |= User('u1', 'Ignored')                              # insert only: u1 exists, nothing changes
jdb &= User('u2', 'Bobby', 26)                            # replace only
jdb['u4'] = User('u4', 'Diana', 28)                       # explicit key

# --- Read: members are filled in place, so one buffer can be reused ---
user = User(id='', name='')
jdb.get_dc('u1', user)
print(user.role, user.address.city, user.balance) # Output: Role.ADMIN Taipei 12.50

# --- Update ---
user.age += 1
jdb += user

# --- Delete ---
jdb -= user
del jdb[User('u2', '')]
del jdb[User('u3', ''), User('u4', '')] # del jdb['u3', 'u4']

# --- Still an ordinary record: queries run against the fields ---
jdb += [User('u5', 'Eve', 41, tags=['python']), User('u6', 'Frank', 22, address=Address('Taipei'))]
print(jdb.find(Query().age > 28))                # Output: {'u5': {'name': 'Eve', 'age': 41, ...}}
print(jdb.find(vals={'address.city': 'Taipei'})) # Output: {'u6': {'name': 'Frank', 'age': 22, ...}}

For explicit control over the file pointer, f_write_dc / f_read_dc are the low-level equivalents of f_write / f_read:

with jdb.open(read_only=False) as fp:
   jdb.f_write_dc(fp, User('u9', 'Grace'), ttl=30)

buf = User(id='', name='')
with jdb.open() as fp:
   for key in jdb.key_table:
      jdb.f_read_dc(fp, key, buf) # filled in place: no allocation per record

Supported field types: nested dataclasses, Optional / Union, Enum, date / time / datetime, Decimal, UUID, Path, List / Dict / Set / Tuple, frozen=True, and init=False fields. Values are rebuilt from the annotations, so a Set[str] comes back as a set even under a data_type whose codec has no native set type.

Unremove & Unmodify

from omni_json_db import JDb
# Initialize the database from file
# Key-Value is Json+Pickle with zstandard compression
jdb = JDb("fruit.jdb", data_type="J+P", zip_type='zs')

# add key
jdb["apple"] = "red"

# modify key
jdb["apple"] = "blue"

# unmodify key (equivalent to jdb.unmodify())
jdb.revert("apple")
assert jdb["apple"] == 'red'

# remove key
del jdb["apple"]
assert "apple" not in jdb

# unremove key (equivalent to jdb.unremove())
jdb.revert("apple")
assert jdb["apple"] == "red"

Backup & Restore

from omni_json_db import JDb
# Initialize the database from file
# Key-Value is MsgPack+Json with Bzip2 compression
jdb = JDb("fruit.jdb", data_type="S+J", zip_type='bz')

# Add fruit to jdb
fruits = {'apple':'red', 'banana':'yellow', 'mango':'yellow', 'lemon':'yellow', 'tomato':'red'}
jdb += fruits
assert jdb == fruits

# backup jdb to bak folder = ./bak/fruit.jdb
jdb_bak = jdb.backup(folder='bak')
assert jdb_bak == jdb

# del all jdb data
del jdb[fruits]
assert len(jdb) == 0

# restore bak folder to jdb
jdb.restore(folder='bak')
assert jdb == fruits

Groups Mode

from omni_json_db import JDb
# Initialize the database from file
# Key-Value is Json+MsgPack with no compression
jdb = JDb('fruit_group.jdb')

# add red group
r_jdb = jdb.add_group('red')
assert r_jdb is jdb['red']

# add yellow group
y_jdb = jdb.add_group('yellow')
assert y_jdb is jdb['yellow']

# add fruits to red group
r_jdb += {'apple': {'qty':1}, 'tomato': {'qty':2}}

# add fruits to yellow group
y_jdb += {'banana': {'qty':4}, 'lemon': {'qty':6}, 'mango': {'qty':8}}

# read group records
print(jdb['red']['apple']['qty'])   # Output: 1
print(jdb['red:::apple'])           # Output: {'red:::apple': {'qty': 1}}
print(jdb['yellow:::banana'])       # Output: {'yellow:::banana': {'qty': 4}}

# find fruits which contain 'a' from all groups
matches = jdb.find(r':::a')
print(matches) # Output: ['red:::apple', 'red:::tomato', 'yellow:::banana', 'yellow:::mango']

Graph Database

omni-json-db natively supports Property Graph structures with the GraphDb class. You can easily manage nodes, edges, and run complex graph algorithms out of the box.

from omni_json_db import GraphDb, Query

# Initialize the graph database in memory (or from a file)
db = GraphDb()

# 1. Add Nodes with Schema-less Properties
db.add_node('Alice', age=25, role='admin')
db.add_node('Bob', age=30, role='user')
db.add_node('Charlie', age=35, role='user')

# 2. Add Edges (Directed or Undirected) with Properties
db.add_edge('Alice', 'Bob', directed=True, weight=1.5, relation='friend')
db.add_edge('Bob', 'Charlie', directed=True, weight=2.0, relation='colleague')
db.add_edge('Charlie', 'Alice', directed=False, weight=0.5) # Undirected edge

# 3. Neighborhood & Adjacency queries (O(1) lookups)
print(db.neighbors('Alice'))
# Output: {'Bob', 'Charlie'}

print(db.degree('Alice'))
# Output: {'in': 0, 'out': 1, 'undirected': 1, 'total': 2}

# 4. Classic Graph Algorithms Built-in
# Find the shortest path using Dijkstra based on edge weights
dist, path = db.dijkstra_shortest_path('Alice', 'Charlie', weight_key='weight')
print(f"Distance: {dist}, Path: {path}")
# Output: Distance: 0.5, Path: ['Alice', 'Charlie']

# Detect cycles in the graph (Alice -> Bob -> Charlie -> Alice)
print(db.is_cyclic())
# Output: True

# 5. Seamless Query Engine Integration
# You can still use the powerful Query object to filter nodes/edges!
q = Query()
admin_nodes = db.find_nodes(q.role == 'admin')
print(list(admin_nodes))
# Output: ['Alice']

# 6. Cascade Deletion
# Removing a node automatically cleans up all connected edges
db.remove_node('Bob')
print(db.has_node('Bob')) # Output: False
print(db.get_edge('Alice', 'Bob', directed=True)) # Output: None

CSV Import / Export

from omni_json_db import JDb
# Initialize the database in memory
# Key-Value is Json+Json with no compression
jdb1 = JDb(data_type="J+J")

# insert value without key
jdb1 += [{'name': 'John', 'age': 22}, {'name': 'John', 'age': 37}, \
         {'name': 'Bob', 'age': 42}, {'name': 'Megan', 'age': 27}]

# export the data to CSV
jdb1.to_csv('example.csv')

# show table
jdb1.show()

# create another JDb in memory
jdb2 = JDb()

# import the data from CSV
jdb2.from_csv('example.csv')
print(jdb2.find(RE='Bob')) # Output: {'2': {'name': 'Bob', 'age': 42}}

# show table
jdb2.show(RE='Bob')

INI / TOML Import

from omni_json_db import JDb
import io

jdb = JDb()

# --- Load INI Format ---
ini_data = """
[server]
host = 127.0.0.1
port = 8080
"""

jdb.from_ini(io.StringIO(ini_data)) # Also supports direct file paths like 'config.ini'
print(jdb['server/host']) # Output: 127.0.0.1

# --- Load TOML Format ---
toml_data = """
app_name = "Omni Test"
[network]
ip = "192.168.1.1"
port = 8181
"""

jdb.from_toml(io.StringIO(toml_data))

print(jdb['/app_name'])    # Output: Omni Test
print(jdb['network/ip'])   # Output: 192.168.1.1

SQLite Import

Step 1: Prepare sample.sqlite

import sqlite3
conn = sqlite3.connect('sample.sql')
cursor = conn.cursor()

cursor.execute('''
CREATE TABLE IF NOT EXISTS projects (
  id INTEGER PRIMARY KEY,
  name text NOT NULL,
  begin_date DATE,
  end_date DATE
)
''')

cursor.execute('''
CREATE TABLE IF NOT EXISTS project_logs (
  project_id INTEGER,
  action TEXT NOT NULL,
  log_date DATE
)
''')

cursor.execute('DELETE FROM projects')
cursor.execute('DELETE FROM project_logs')

projects_data = [
  (1, 'cooking', '2000-01-02', '2003-01-13'),
  (2, 'reading', '2023-05-01', '2023-12-31'),
  (3, 'coding', '2024-01-01', '2024-06-30')
]
cursor.executemany('INSERT INTO projects (id, name, begin_date, end_date) VALUES (?, ?, ?, ?)', projects_data)

logs_data = [
  (1, 'bought ingredients', '2000-01-01'),
  (1, 'started cooking', '2000-01-02'),
  (2, 'bought books', '2023-04-20'),
  (3, 'setup environment', '2024-01-01')
]
cursor.executemany('INSERT INTO project_logs (project_id, action, log_date) VALUES (?, ?, ?)', logs_data)

conn.commit()
conn.close()

Step 2: Import to JDb

from omni_json_db import JDb
jdb = JDb("migrated_data.jdb")

# Load an entire SQLite database with one line of code
jdb.from_sqlite('sample.sqlite')

# SQLite tables (e.g., 'projects' and 'project_logs') automatically become groups
projects = jdb['projects']
logs = jdb['project_logs']

# Query relational data using the NoSQL interface
print(projects[3]['name'])  # Get the name of the project with ID 3
print(len(logs))            # Get the total number of logs

# Combine with powerful Lambda queries to find logs for a specific project
project_3_logs = logs.find(FUNC=lambda val: val['project_id'] == 3)

Parquet Import

Requires the optional pyarrow dependency (pip install pyarrow).

import pyarrow as pa
import pyarrow.parquet as pq

# write a sample Parquet file
table = pa.table({
    '_id':   ['p1', 'p2', 'p3'],
    'name':  ['Alice', 'Bob', 'Charlie'],
    'age':   [30, 25, 35],
})
pq.write_table(table, 'sample.parquet')

from omni_json_db import JDb
jdb = JDb()

# by default, the first column ('_id') becomes the record key
jdb.from_parquet('sample.parquet')
print(jdb['p1'])  # Output: {'name': 'Alice', 'age': 30}

# rows stream in via pyarrow RecordBatches, so multi-GB files
# import with a small, constant memory footprint
jdb.from_parquet('sample.parquet', batch_size=65536)

# pick a different key column, and prune to only the columns you need
jdb.from_parquet('sample.parquet', key='_id', columns=['name'])

Network Mode

Server side

from omni_json_db import JDb, run_files_server

jdb = JDb('storage.jdb')

# equivalent to: files='storage.jdb'
run_files_server(host='127.0.0.1', port=59898, files=jdb)

# write key to JDb
jdb['remote-key'] = 'secret'

Client side

from omni_json_db import JDb

# connect to files server
jdb = JDb('127.0.0.1:59898')

# read remote key from JDb
print(jdb['remote-key']) # Output: secret

Change Type

from omni_json_db import JDb

# Initialize the database in memory
# Key-Value is Json+Json with no compression
jdb = JDb(data_type='J+J')

fruits = {'apple':'red', 'banana':'yellow', 'mango':'yellow', 'lemon':'yellow', 'tomato':'red'}

# add all fruits to database
jdb += fruits
assert jdb == fruits
print(jdb.data_type, jdb.zip_type) # Output: J+J no

# change data_type to 'S+S' and zip_type to 'lz'
jdb.upgrade(data_type='S+S', zip_type='lz')
assert jdb == fruits
print(jdb.data_type, jdb.zip_type) # Output: S+S lz

# only change KEY type from 'S' to 'J'
jdb.change_KEY('J')
assert jdb == fruits
print(jdb.data_type, jdb.zip_type) # Output: J+S lz

Time-Series

from omni_json_db import JDb
import datetime as dt

# Initialize the database in memory
# Key+Value is Json+Json with Gzip compression
# using BTree as Key Table for better memory usage
jdb = JDb(data_type="J+J(gz)", key_limit="bt")

# insert data
fruits = {'apple':'red', 'banana':'yellow', 'mango':'yellow', 'lemon':'yellow', 'tomato':'red'}
jdb += fruits

# datetime for create date, date for modify date
now = dt.datetime.now()
today = now.date()

# find create date: date == now
matches = jdb[now]
assert matches == fruits

# find create date: date >= now
matches = jdb[now:]
assert matches == fruits

# find create date: date < now
matches = jdb[:now]
assert len(matches) == 0

# find create date: now <= date <= now+1
next_date = now + dt.timedelta(days=1)
matches = jdb[now:next_date]
assert matches == fruits

prev_date = now - dt.timedelta(days=1)
prev_week = now - dt.timedelta(days=7)

# change key create date
jdb.keys['apple', 'tomato'] = prev_date
jdb.keys['mango'] = prev_week
assert jdb[prev_date] == {'apple':'red', 'tomato':'red'}
assert jdb[prev_week] == {'mango':'yellow'}

# find create date: date == now
matches = jdb[now]
assert set(matches) == {'banana', 'lemon'}

# find create date: date < now
matches = jdb[:now]
assert set(matches) == {'apple', 'mango', 'tomato'}

# find modify date: date == today
matches = jdb[today]
assert matches == fruits

# change key modify date + create date
new_modify_date = prev_date.date()
new_create_date = prev_week.date()
assert new_modify_date >= new_create_date
jdb.keys['lemon'] = f'{new_modify_date} {new_create_date}'

# find modify date: date == today
matches = jdb[today]
assert set(matches) == {'apple', 'banana', 'mango', 'tomato'}

# find modify date: date == prev_date
matches = jdb[prev_date.date()]
assert set(matches) == {'lemon'}

# change all keys create date
jdb.keys[:] = today
assert jdb[today] == fruits

Operator

from omni_json_db import JDb
# Initialize the database in memory
# Key+Value is MsgPack+MsgPack with lz4 compression
jdb = JDb(data_type="S+S(lz)")

# [1] KEY+VAL operators
# <jdb += data> == jdb.update(data)
data = {f'key{v}':v for v in range(100)}
jdb += data
assert len(jdb) == 100

# <jdb == data>
assert jdb == data

# <jdb |= ..> == jdb.insert(..)
jdb |= {f'key{v}':v+1 for v in range(102)}
assert jdb['key100'] == 101
assert jdb[-2.:] == {'key100':101, 'key101':102} # get last two modified records
assert jdb[(f'key{v}' for v in range(100))] == data # equivalent to jdb[data] == data

# <jdb -= ..> == jdb.remove(..)
jdb -= ['key100', 'key101', 'key102', 'key103']
assert jdb == data

# <jdb &= ..> == jdb.replace(..)
jdb &= {f'key{v}':v+1 for v in range(200)}
assert jdb == {f'key{v}':v+1 for v in range(100)}

# <jdb ^= ..> == jdb.unmodify(..)
jdb ^= {f'key{v}' for v in range(100)} # equivalent to jdb ^= data
assert jdb == data

# <jdb[:] = ..> == jdb.update(..)
jdb[:] = 0 # set all records to zero
assert jdb == {f'key{v}':0 for v in range(100)}
assert jdb.find(NE=0) == {}

# remove all records
jdb -= jdb # equivalent to del jdb[:]
assert len(jdb) == 0

# <jdb ^= ..> == jdb.unremove(..)
jdb ^= {f'key{v}' for v in range(100)} # equivalent to jdb ^= data
assert all(val == 0 for key,val in jdb.items())

# lambda VALUE operation
jdb[:] = lambda key,val: int(key.replace('key', '')) + val
assert jdb == data

# <del jdb[..]> == jdb.remove_fast(..)
del jdb[data] # equivalent to del jdb[:]

# unremove all data
jdb ^= data
assert jdb == data

# <jdb[..]> == jdb.get_n(..) or jdb.get_all()
matches = jdb[('key2', 'key22', 'key44', 'key111')]
assert matches == {'key2':2, 'key22':22, 'key44':44}

# lambda KEY operation
matches = jdb[lambda key:key.endswith('1')]
assert set(matches) == {'key1', 'key11', 'key21', 'key31', 'key41', 'key51', 'key61', 'key71', 'key81', 'key91'}

# set all matched records to -1
jdb[matches] = -1
matches_2 = jdb[lambda key,val: val == -1]
assert set(matches) == set(matches_2)
assert matches_2 == jdb.find(EQ=-1)
assert matches_2 == jdb.find(FUNC=lambda val: val == -1)

# RE search
matches_3 = jdb[::r'1$']
assert matches_2 == matches_3

# unmodify
jdb ^= matches
assert jdb == data

# [2] KEY operators
# <jdb & {..}> == jdb.intersection(..)
matches = jdb & {f'key{v}' for v in range(98, 120)}
assert matches == {'key98', 'key99'}

# <{..} & jdb> == {..}.intersection(jdb)
matches_2 = {f'key{v}' for v in range(98, 120)} & jdb
assert matches == matches_2

# <jdb | {..}> == jdb.union(..)
matches = jdb | {f'key{v}' for v in range(10, 120)}
assert matches == {f'key{v}' for v in range(0, 120)}

# <{..} | jdb> == {..}.union(jdb)
matches_2 = {f'key{v}' for v in range(10, 120)} | jdb
assert matches == matches_2

# <jdb + {..}> == jdb.union(..)
matches = jdb + {f'key{v}' for v in range(10, 120)}
assert matches == matches_2

# <{..} + jdb> == {..}.union(jdb)
matches_2 = {f'key{v}' for v in range(10, 120)} + jdb
assert matches == matches_2

# <jdb - {..}> == jdb.difference(..)
matches = jdb - {f'key{v}' for v in range(0, 98)}
assert matches == {'key98', 'key99'}

# <{..} - jdb> == {..}.difference(jdb)
matches = {f'key{v}' for v in range(2, 102)} - jdb
assert matches == {'key100', 'key101'}

# <jdb ^ {..}> == jdb.non_intersection(..)
matches = jdb ^ {f'key{v}' for v in range(1, 101)}
assert matches == {'key0', 'key100'}

# <{..} ^ jdb> == {..}.non_intersection(jdb)
matches_2 = {f'key{v}' for v in range(1, 101)} ^ jdb
assert matches == matches_2

# <.. in jdb> == jdb.has_all(..)
assert 'key10' in jdb
assert {'key10', 'key90'} in jdb
assert {'key10', 'key90', 'key110', 'key190'} not in jdb
assert jdb.has('key10')
assert jdb.has_all('key10')
assert jdb.has_any('key10')
assert jdb.has_all({'key10', 'key90'})
assert jdb.has_any({'key10', 'key90', 'key110', 'key190'})
assert jdb.is_disjoint({'key110', 'key190'})

All standard set methods work: union(), intersection(), difference(), isdisjoint(), issubset(), issuperset().

More Query Examples

Below are examples of how to utilize the various parameters and NoSQL syntax.

from omni_json_db import JDb
import re

# Initialize an in-memory database
jdb = JDb()

# Sample user records
users = {
   'user_1': {'name': 'Alice', 'age': 30, 'email': 'alice@example.com', 'role': 'admin', 'tags': ['python', 'database']},
   'user_2': {'name': 'Bob', 'age': 25, 'role': 'developer', 'tags': ['javascript', 'web']},
   'user_3': {'name': 'Charlie', 'age': 35, 'role': 'developer', 'tags': ['python', 'linux', 'aws']},
   'user_4': {'name': 'Diana', 'age': 28, 'email': 'diana@test.com', 'role': 'designer', 'tags': ['ui', 'ux']}
}

# Insert data
jdb += users

# 1. Exact Match & Global Search (ANY, RE, RE2)
#----------------------------------------------------------
# Find users where any attribute exactly matches 'Alice'
res = jdb.find(ANY='Alice')
assert list(res) == ['user_1']

res2 = jdb.find(vals={'name': 'Alice'})
assert res == res2

# converts values into JSON string format for searching.
# Find any record that has the string 'designer' inside it
res = jdb.find(RE=r'designer') # vals={'$re':r'designer'}
assert list(res) == ['user_4']

# RE2 removes JSON symbols (,[]{}") before searching
res = jdb.find(RE2=r'role:designer')
assert list(res) == ['user_4']

res2 = jdb.show(vals={'role.$re': r'd.+ner'}) # {'role': re.compile(r'd.+ner')}
assert res == res2

# 2. Relational & Conditional Operators (vals)
#----------------------------------------------------------
# Age is greater than or equal to 30
res = jdb.find(vals={'age': {'$gte': 30}}) # find(ANY={'$gte': 30})
assert list(res) == ['user_1', 'user_3']

res2 = jdb.find(vals={'age.$ge': 30})
assert res == res2

# Age is strictly less than 30
res = jdb.find(vals={'age': {'$lt': 30}}) # find(ANY={'$lt': 30})
assert list(res) == ['user_2', 'user_4']

res2 = jdb.find(vals={'age.$lt': 30})
assert res == res2

# Role is either 'admin' or 'designer'
res = jdb.find(vals={'role': {'$in': ['admin', 'designer']}})
assert list(res) == ['user_1', 'user_4']

res2 = jdb.find(vals={'role': ['admin', 'designer']}) # {'role.$in': ['admin', 'designer']}
assert res == res2

# tags contains 'python'
res = jdb.find(vals={'tags': {'$has': 'python'}})
assert list(res) == ['user_1', 'user_3']

res2 = jdb.find(vals={'tags.$has': 'python'})
assert res == res2

# Age is NOT 30
res = jdb.find(vals={'age': {'$ne': 30}}) # find(ANY={'$ne': 30})
assert list(res) == ['user_2', 'user_3', 'user_4']

res2 = jdb.find(vals={'!age': 30}) # {'age.$ne': 30}
assert res == res2

# Age is 28
res = jdb.find(vals={'age': {'$eq': 28}}) # find(ANY={'$eq': 28})
assert list(res) == ['user_4']

res2 = jdb.find(vals={'age': 28})
assert res == res2

# 40 >= Age > 25
res = jdb.find(vals={'age': {'$gt': 25, '$lte': 40}}) # {'age.$gt':25, 'age.$lte':40}
assert list(res) == ['user_1', 'user_3', 'user_4']

res2 = jdb.show(vals={'age.$between': (26, 40)})
assert res == res2

# 3. Logical Grouping (AND, OR, NOR, NOT)
#----------------------------------------------------------
# Age >= 25 AND Age <= 30
res = jdb.find(AND=[{'age': {'$gte': 25}}, {'age': {'$lte': 30}}])
assert list(res) == ['user_1', 'user_2', 'user_4']

res2 = jdb.find(vals={'age.$ge': 25, 'age.$le': 30})
assert res == res2

# Role is 'admin' OR Age > 30
res = jdb.find(OR=[{'role': 'admin'}, {'age': {'$gt': 30}}])
assert list(res) == ['user_1', 'user_3']

res2 = jdb.find(OR=[{'r*e': 'admin'}, {'age.$gt': 30}])
assert res == res2

# Role is not 'admin' AND Age <= 30
res = jdb.find(NOR=[{'role': 'admin'}, {'age.$gt': 30}])
assert list(res) == ['user_2', 'user_4']

# User is NOT a developer
res = jdb.find(NOT={'role': 'developer'})
assert list(res) == ['user_1', 'user_4']

res2 = jdb.find(vals={'!role': 'developer'})
assert res == res2

# (Role is 'admin' OR Age > 30) AND 'linux' not in tags
res = jdb.find(AND=[
   {'$or': [
      {'role': 'admin'},
      {'age': {'$gt': 30}}
   ]},
   {'$not': {'tags': {'$has': 'linux'}}}
])
assert list(res) == ['user_1']

res2 = jdb.show(vals={'$or': [{'role': 'admin'}, {'age.$gt': 30}], '!tags.$has': 'linux'})
assert res == res2

# 4. Regular Expressions (RE, RE2, re.compile)
#----------------------------------------------------------
# Values matching an email domain regex
res = jdb.find(vals={'email': re.compile(r'.@example.com')})
assert list(res) == ['user_1']

# Find users where any attribute exactly matches regex
res = jdb.find(ANY=re.compile(r'.@example.com'))
assert list(res) == ['user_1']

# Global regex search for strings containing 'li' (matches 'Alice', 'Charlie', 'linux')
res = jdb.find(RE=r'li[a-z]')
assert list(res) == ['user_1', 'user_3']

# Match specific Database Keys using compiled regex (e.g., matching 'user_1', 'user_2')
res = jdb.show(re.compile(r'^user_[1-2]$'))
assert list(res) == ['user_1', 'user_2']

# 5. Array / List Operations
#----------------------------------------------------------
# Users with exactly 2 tags in their list
res = jdb.find(vals={'tags': {'$size': 2}})
assert list(res) == ['user_1', 'user_2', 'user_4']

res2 = jdb.find(vals={'tags.$size': 2})
assert res == res2

# Users whose FIRST tag (index 0) is 'python'
res = jdb.find(vals={'tags': {'$0': 'python'}})
assert list(res) == ['user_1', 'user_3']

res2 = jdb.show(vals={'tags.0': 'python'})
assert res == res2

# 6. Lambda / Custom Functions (FUNC) & Pagination (limit)
#----------------------------------------------------------
# Pass a lambda to evaluate both the key and the value dynamically
# Example: Find the first users whose age is an even number
res = jdb.find(
    FUNC=lambda k, v: isinstance(v, dict) and v.get('age', 1) % 2 == 0,
   limit=1
)
assert list(res) == ['user_1']

res2 = jdb.find(vals={'age.$mod': (2, 0)}, limit=1)
assert res == res2

# Users with email
res = jdb.find(vals={'email': lambda v: v != ''})
assert list(res) == ['user_1', 'user_4']

res2 = jdb.find(EXISTS='email')
assert res == res2

# Users without email
res = jdb.find(NOT={'email': lambda v: v != ''})
assert list(res) == ['user_2', 'user_3']

res2 = jdb.find(vals={'!$exists': 'email'})
assert res == res2

# For primitive stored values (non-nested), you can use quick keyword arguments:
jdb['simple_counter'] = 50
res = jdb.find(EQ=50)
assert list(res) == ['simple_counter']

res = jdb.show(IN=[40, 50]) # Value in list
assert list(res) == ['simple_counter']

Pythonic Query Examples

For developers who prefer a Pythonic and object-oriented syntax for filtering data (similar to the TinyDB experience), omni-json-db provides the Query object. You can elegantly use native Python operators (e.g., ==, >, &, |, ~) and chained methods to construct complex search conditions.

from omni_json_db import JDb, Query

# 1. Initialize database and add test data
jdb = JDb()
jdb += {
    'user_1': {'name': 'Alice', 'age': 30, 'email': 'alice@example.com', 'role': 'admin', 'tags': ['python', 'database']},
    'user_2': {'name': 'Bob', 'age': 25, 'role': 'developer', 'tags': ['javascript', 'web']},
    'user_3': {'name': 'Charlie', 'age': 35, 'role': 'developer', 'tags': ['python', 'linux', 'aws']},
    'user_4': {'name': 'Diana', 'age': 28, 'email': 'diana@test.com', 'role': 'designer', 'tags': ['ui', 'ux']}
}

# 2. Create a Query instance
User = Query()

# Basic Comparison: Find users older than 28
res = jdb.find(User.age > 28)
assert set(res) == {'user_1', 'user_3'}

# Logical Combinations (AND & OR): Find developers under 30 OR admins
res = jdb.find((User.role == 'developer') & (User.age < 30) | (User.role == 'admin'))
assert set(res) == {'user_1', 'user_2'}

# Array Query: Find users whose tags include 'python'
res = jdb.find(User.tags.has('python'))
assert set(res) == {'user_1', 'user_3'}

# Path Wildcard: Regex search across all fields recursively (Find email containing example.com)
res = jdb.find(User['**'].matches(r'.@example\.com'))
assert set(res) == {'user_1'}

# Advanced Filters: Find users who DO NOT have an 'email' field (~ is the NOT operator)
res = jdb.find(~User.exists('email'))
assert set(res) == {'user_2', 'user_3'}

# Lambda Test: Find users whose age is an even number
res = jdb.find(User.age.test(lambda age: age % 2 == 0), sort=User.name, reverse=True)
assert set(res) == {'user_1', 'user_4'}

# rename admin to Administrator
res = jdb.update_if(User.role == 'admin', {'role': 'Administrator'})
assert res == 1 and (User.role == 'Administrator') in jdb

# sorted by user name
res = jdb.show(sort=User.name, reverse=True)
assert res == jdb

# grouped by user role
res = jdb.show(group_by=User.role) # jdb.show(group_by='role')
assert 'developer' in res

res = jdb.show(group_by={'role':['name', 'age.$avg']})
assert 'developer' in res

Record Flags

Every record carries a set of JKeyFlag bits in the key index, giving it file-system-like attributes. Set them with a chmod-style string: a bare letter or + sets a flag, - clears it, and anything you don’t name keeps its current value.

Flags come in three kinds. Stored flags are properties of the record and persist on disk. Derived flags are maintained by the engine and cannot be set by name. Transient flags describe a single call rather than a record — they are never written to disk, and passing one leaves the record’s stored flags untouched.

Stored — properties of the record

READ_ONLY and APPEND_ONLY block deleting as well as writing, so NO_DELETE is the only way to say this record may change, but must never disappear — the retention guarantee a config row or an account record wants.

Derived — maintained by the engine

Transient — properties of a single call

Transient flags are passed to the operation itself, not to set_key_flags(). They are consumed rather than stored, so jdb.set(k, v, key_flags='w+r') refuses an existing key and stores READ_ONLY on a new one. Naming a transient flag in a find() predicate raises JValueError rather than silently matching every record.

from omni_json_db import JDb, JKeyFlag

jdb = JDb()
jdb['config']        = {'theme': 'dark'}
jdb['audit/2026-08'] = ['created']

# read-only: writes are refused, silently
jdb.set_key_flags('config', 'r')
jdb['config'] = {'theme': 'light'}
print(jdb['config'])                    # Output: {'theme': 'dark'}

# append-only + no history: an audit log that can only grow
jdb.set_key_flags('audit/2026-08', 'av')
jdb['audit/2026-08'] = ['created', 'updated']   # extension -> allowed
jdb['audit/2026-08'] = ['wiped']                # not an extension -> refused
del jdb['audit/2026-08']                        # refused too
print(jdb['audit/2026-08'])             # Output: ['created', 'updated']

# inspect, then unlock again ('-r' clears only READ_ONLY).
# every flag API returns {key: (flags, ttl)}
flags, ttl = jdb.get_key_flags('audit/2026-08')['audit/2026-08']
print(str(JKeyFlag(flags)), ttl)        # Output: __a_v_____________ 0
jdb.set_key_flags('config', '-r')
jdb['config'] = {'theme': 'light'}
print(jdb['config'])                    # Output: {'theme': 'light'}

# hidden: kept out of query results, but not a secret — items() still sees it
jdb['_scratch'] = 'internal'
jdb.set_key_flags('_scratch', 'h')
print(list(jdb.find()))                 # Output: ['config', 'audit/2026-08']
print(jdb['_scratch'])                  # Output: internal
print(list(jdb.find(with_hidden=True))) # Output: ['config', 'audit/2026-08', '_scratch']

# tag records for your own use, and combine flags freely
jdb.set_key_flags('config', '+0')       # USER0
jdb.keys.set_flags('config', '+c-0')    # add NO_CACHE, drop USER0

# no-delete: the record stays editable, but nothing can remove it
jdb['account/42'] = {'plan': 'pro'}
jdb.set_key_flags('account/42', '+p')
jdb['account/42'] = {'plan': 'team'}    # update -> allowed
print(jdb['account/42'])                # Output: {'plan': 'team'}
print(jdb.remove('account/42'))         # Output: {}
print(jdb.remove('account/42', key_flags='u'))
                                        # Output: {'account/42': {'plan': 'team'}}

# create-only ('w') and update-only ('m'), decided under the write handle
# rather than by a test-then-write, so neither races another writer
print(jdb.set('config', {'theme': 'dark'}, key_flags='w'))   # exists -> refused
                                        # Output: None
print(jdb.set('session/abc', ['new'], key_flags='w'))
                                        # Output: ['new']
print(jdb.set('session/xyz', ['nope'], key_flags='m'))       # absent -> refused
                                        # Output: None
print(jdb.set('session/abc', ['touched'], key_flags='m'))
                                        # Output: ['touched']

# no-follow: read the link itself instead of what it points at
jdb.set_link('latest', 'config')
print(jdb['latest'])                    # Output: {'theme': 'light'}
print(jdb.get('latest', key_flags='n')) # Output: config