KivoDB — multi-driver database toolkit for Python

KivoDB is a standalone Python database toolkit inspired by the public API of good-db/good.db, a TypeScript/Node.js key/value database wrapper. KivoDB has its own name and package identity; it preserves the compatible camelCase method surface, nested-key model, driver contract, array/math/collection helpers, cache options, table proxy, and Convertor concept.

Project status: first Python distribution (0.1.0), API compatibility target: upstream good.db 2.5.0. This is an unofficial independent implementation, not published, affiliated with, or maintained by the upstream authors. The documented public methods are implemented and covered by the local test suite; live connections to MongoDB, PostgreSQL, and MySQL need to be verified against services you operate. See COMPATIBILITY.md.

Features

• Familiar API: KivoDB, set, get, delete, push, find, add, all, table, and the other public helpers from upstream.
• Multiple storage backends: memory, JSON, YAML, SQLite, MongoDB, PostgreSQL, and MySQL.
• Same style of API for sync and async drivers: sync drivers return values immediately; async drivers return awaitables.
• Nested keys using a configurable separator, disabled by default.
• Optional in-process LRU cache.
• Table/collection proxies and conversion between drivers.
• No mandatory third-party dependencies. The standard-library drivers only need Python itself.
• Type-checker marker (py.typed) and tests included in the source distribution.

Package and import names

The PyPI distribution name and Python import root are both kivodb. The primary class is KivoDB; GoodDB is also exported as a compatibility alias for code using the upstream API name. KivoDB is independently named and is not affiliated with the upstream project.

Installation

After the distribution has been published to PyPI:

python -m pip install kivodb

Install optional drivers only as needed:

python -m pip install 'kivodb[yaml]'        # YMLDriver (PyYAML)
python -m pip install 'kivodb[mongodb]'     # MongoDBDriver (PyMongo Async API)
python -m pip install 'kivodb[postgresql]'  # PostgreSQLDriver (asyncpg)
python -m pip install 'kivodb[mysql]'       # MySQLDriver (aiomysql)
python -m pip install 'kivodb[all]'         # all optional drivers

To install this checkout locally:

python -m pip install .

For development and tests:

python -m pip install -e '.[dev,all]'
python -m pytest

Quick start

JSON file

from kivodb import KivoDB, JSONDriver

db = KivoDB(
    JSONDriver({"path": "./database.json", "format": True}),
    {
        "table": "data",
        "nested": "..",
        "nestedIsEnabled": True,
        "cache": {"isEnabled": True, "capacity": 1024},
    },
)

db.set("users..123..profile", {"name": "Abdallah", "level": 1})
db.set("users..123..coins", 500)

db.add("users..123..coins", 100)
print(db.get("users..123..coins"))  # 600
print(db.get("users..123..profile"))
print(db.all())

SQLite (no extra dependency)

from kivodb import KivoDB, SQLiteDriver

db = KivoDB(SQLiteDriver({"path": "./database.sqlite"}), {"table": "data"})
db.set("user:123", {"name": "Abdallah", "coins": 500})
print(db.get("user:123"))
db.driver.close()  # close the underlying sqlite3 connection when finished

Async MongoDB

import asyncio
from kivodb import KivoDB, MongoDBDriver

async def main():
    db = KivoDB(
        MongoDBDriver({
            "uri": "mongodb://127.0.0.1:27017",
            "database": "example",
        }),
        {"table": "data", "nested": ".", "nestedIsEnabled": True},
    )
    await db.connect()
    try:
        await db.set("user:123", {"name": "Abdallah", "coins": 500})
        print(await db.get("user:123"))
        await db.add("user:123.coins", 100)  # nested keys must be enabled first
    finally:
        await db.disconnect()

asyncio.run(main())

The example above shows the async lifecycle. To use nested dot keys, configure them on KivoDB:

db = KivoDB(driver, {
    "table": "data",
    "nested": ".",
    "nestedIsEnabled": True,
})

Do not use asyncio.run() inside an already-running event loop (for example, inside an async Discord bot event handler); call await on the methods instead.

Defaults and configuration

KivoDB(driver=None, options=None)

Option	Default	Meaning
table	"gooddb"	Table/collection for this KivoDB instance
nested	".."	Separator for nested keys
nestedIsEnabled	False	Enables nested-key parsing
cache.isEnabled	False	Enables the in-process LRU cache
cache.capacity	1024	Maximum number of cached root keys

With no driver, KivoDB() creates SQLiteDriver({"path": "./database.sqlite"}). Calling SQLiteDriver() directly uses ./db.sqlite. JSONDriver() defaults to ./db.json; YMLDriver() defaults to ./db.yml; both can be overridden.

When nestedIsEnabled=True, for example db.set("users..123..coins", 500), KivoDB stores a root record under key users and modifies the nested structure inside it. The separator can be changed, e.g. to ".". If nested keys are disabled, a key containing .. is treated as a literal key.

A method-level options dictionary can override the nested settings for one call:

# Nested parsing disabled for this particular operation:
db.set("user..name", "literal value", {})

Drivers

Driver	API mode	Optional dependency	Default path / notes
MemoryDriver	Sync	None	In-process dictionary; contents disappear when the process exits
CacheDriver	Sync	None	Legacy alias for MemoryDriver; emits DeprecationWarning
JSONDriver	Sync	None	./db.json; whole file is read/written per operation
YMLDriver	Sync	PyYAML	./db.yml; whole file is read/written per operation
SQLiteDriver	Sync	None	./db.sqlite; one JSON-encoded value per row
MongoDBDriver	Async	pymongo	Native PyMongo AsyncMongoClient API
PostgreSQLDriver	Async	asyncpg	Uses an async connection pool and JSONB values
MySQLDriver	Async	aiomysql	Uses an async connection pool and JSON-encoded values

Driver option examples

JSONDriver({"path": "./database.json", "format": True})
YMLDriver({"path": "./database.yml"})
SQLiteDriver({"path": "./database.sqlite"})
MongoDBDriver({"uri": "mongodb://localhost:27017", "database": "my_app"})
PostgreSQLDriver({"user": "app", "password": "...", "database": "my_app", "host": "127.0.0.1"})
MySQLDriver({"user": "app", "password": "...", "db": "my_app", "host": "127.0.0.1"})

The SQL drivers treat table as a SQL identifier, quote it, and use parameter placeholders for values. Database user permissions still control which tables can be created/read/written. Keep connection credentials out of source code; load them from environment variables or a secret manager.

Sync vs async method calls

For MemoryDriver, JSONDriver, YMLDriver, and SQLiteDriver, call methods normally:

db.set("visits", 1)
visits = db.get("visits")

For MongoDBDriver, PostgreSQLDriver, and MySQLDriver, connect first and await database operations:

await db.connect()
await db.set("visits", 1)
visits = await db.get("visits")
await db.disconnect()

db.isAsync identifies the driver's mode. Do not forget to await operations on async drivers; otherwise their coroutines will not execute.

API reference

The camelCase names intentionally match the TypeScript library. All operations below accept optional options where indicated in the code signatures. Async drivers make the same operations awaitable.

Database methods

Method	Description	Return value
set(key, value, options=None)	Create or replace a value	True
get(key, options=None)	Read a value; missing keys return None in Python	Value or None
delete(key, options=None)	Delete a key	True
setMany(data, options=None)	Set a non-empty mapping of key/value pairs	True
getMany(keys, options=None)	Read multiple keys into a mapping	dict
deleteMany(keys, options=None)	Delete multiple keys	True
has(key, options=None)	Check whether a key is present, with upstream truthiness behavior for sync/async drivers	bool

setMany currently iterates through keys using the public set method; it is not guaranteed to be a single backend bulk transaction.

Array methods

These operate on a list saved at key and raise DatabaseError when the stored value is not an array (subject to the missing-value behavior shown below).

Method	Description
push(key, value, options=None)	Append an item and return the new length
unshift(key, value, options=None)	Insert an item at the beginning and return the new length
pop(key, options=None)	Remove and return the last item
shift(key, options=None)	Remove and return the first item
pull(key, valueOrCallback, pullAll=False, options=None)	Remove matching item(s); matching may be a value or predicate callback
find(key, callback, options=None)	Return the first item satisfying a callback, or None
filter(key, callback, options=None)	Return all items satisfying a callback
findAndUpdate(key, findCallback, updateCallback, options=None)	Replace the first match with the update callback's return value
findAndUpdateMany(key, findCallback, updateCallback, options=None)	Replace all matches and return the updated items
distinct(key, value=_MISSING, options=None)	Remove duplicate values by default; passing a callback follows upstream filter behavior

Callbacks accept one, two, or three positional arguments: (value), (value, index), or (value, index, array). Callback bodies should be synchronous and return their result. For findAndUpdate / findAndUpdateMany, return the updated object/value from the update callback; mutating an object alone without returning it is not enough.

Math methods

Method	Description
add(key, value, options=None)	Add to the current number (missing/falsy baseline follows upstream semantics)
subtract(key, value, options=None)	Subtract from the current number
multiply(key, value, options=None)	Multiply the current number
double(key, options=None)	Double the current number
math(key, mathSign, value, options=None)	Apply +, -, *, ×, or /

Division is called using db.math("score", "/", 2). The upstream class does not expose a separate divide() method. Dividing by zero raises DatabaseError.

Collection methods

Method	Description
startsWith(key, options=None)	Return entries whose keys start with the given text
endsWith(key, options=None)	Return entries whose keys end with the given text
includes(key, options=None)	Return entries whose keys contain the given text
keys()	List keys from the current table
values()	List values from the current table
all(type="object")	Return all records as a mapping; all("array") returns [{"key": ..., "value": ...}]
clear()	Remove all records in the current table
type(key, options=None)	Return null, boolean, number, string, array, object, or unknown
size(key, options=None)	Length of strings, lists, and dictionaries; otherwise 0
table(name)	Return another KivoDB instance bound to the requested table
connect() / disconnect()	Open/close an async driver; unsupported on sync drivers

The startsWith, endsWith, and includes operations search record keys, not the contents of stored strings. With nested-key options enabled, passing a nested path searches keys within the selected parent object.

Table proxies

A table proxy shares the same driver but changes the table name:

users = db.table("users")
users.set("123", {"name": "Sam"})
print(users.get("123"))

For async drivers, table creation is awaitable:

users = await db.table("users")
await users.set("123", {"name": "Sam"})

LRU cache

db = KivoDB(
    SQLiteDriver({"path": "./database.sqlite"}),
    {"table": "data", "cache": {"isEnabled": True, "capacity": 2048}},
)

The cache is an in-process LRU cache, not Redis or a shared cache. Each process has its own cache; it is not shared between bot shards, workers, or machines. Use cache.isEnabled=False for cache-free reads. clear() invalidates the instance's local cache, but data changed through another independent KivoDB instance or external client may remain stale until cache eviction. If many instances write the same table, disable caching unless you handle invalidation yourself.

Convert data between drivers

Convertor can copy one table or all tables from one driver to another. Its convert() method is asynchronous even when both drivers are synchronous.

import asyncio
from kivodb import Convertor, JSONDriver, SQLiteDriver

async def main():
    convertor = Convertor({
        "from": JSONDriver({"path": "./old.json"}),
        "to": SQLiteDriver({"path": "./new.sqlite"}),
        "table": "data",  # use "all_tables" to copy every table
    })
    await convertor.convert()

asyncio.run(main())

Conversion is not a distributed transaction. Back up source and target data first, especially for large or live production databases.

Error handling

DatabaseError is raised for invalid keys, unsupported operations, incorrect array types, and invalid math operations. Driver-level connection and filesystem failures may raise exceptions from the relevant standard library or optional driver package.

from kivodb import DatabaseError, KivoDB, MemoryDriver

db = KivoDB(MemoryDriver())
try:
    db.set(" ", 1)
except DatabaseError as exc:
    print(f"Invalid database operation: {exc}")

Compatibility notes

The goal is to preserve the public API and data model, not to emulate every edge case of JavaScript's runtime. In particular:

• JavaScript distinguishes undefined from null; Python callers receive None for a missing value.
• Sync has() follows upstream JavaScript truthiness. Values such as 0, False, and "" therefore return False; async has() tests for a non-None value, matching the current upstream implementation's difference.
• The Python port corrects implementation hazards where appropriate (including cache invalidation on clear() and the driver contract for MemoryDriver.getAllRows()).
• The MongoDB implementation uses PyMongo's async API; it does not use Motor.
• SQL identifiers are quoted and query values are parameterized where applicable.

See COMPATIBILITY.md for the full scope and known differences.

Development and releases

See CONTRIBUTING.md for the development setup. Standard release artifacts are built with:

python -m pip install --upgrade build twine
python -m pytest
python -m build
python -m twine check dist/*

The GitHub Actions release workflow is configured for PyPI Trusted Publishing after you create the repository and configure its trusted publisher on PyPI. This project bundle cannot upload to PyPI by itself; publishing requires an account/project owner to authorize the release.

License and attribution

MIT. See LICENSE and NOTICE. Upstream API reference: good-db/good.db.