ScriptDB is a tiny wrapper around SQLite with built‑in migration support. It can be used asynchronously or synchronously. ScriptDB is designed for small integration scripts and ETL jobs where using an external database would be unnecessary. The project aims to provide a pleasant developer experience while keeping the API minimal.
- Async and sync – choose between the async
aiosqlitebackend or the synchronous stdlibsqlite3backend. - Migrations – declare migrations as SQL snippet(s) or Python callables and let ScriptDB apply them once.
- Lightweight – no server to run and no complicated setup; perfect for throw‑away scripts or small tools.
- WAL by default – connections use SQLite's write-ahead logging mode;
disable with
use_wal=Falseif rollback journals are required.
Composite primary keys are not supported; each table must have a single-column primary key.
ScriptDB requires SQLite 3.24.0 or newer. Most modern Python builds ship with a recent-enough SQLite; on older
distributions (e.g., Ubuntu 18.04) installing scriptdb[pysqlite] bundles a modern SQLite build. If your environment
only provides an older SQLite, the compatibility layer will emit a warning and upsert_one/upsert_many will raise a
clear error until you upgrade.
If pysqlite still does not help in your environment, you can open the database with
legacy_sqlite_support=True. In that mode ScriptDB emulates upserts using older SQLite-compatible SQL. It keeps
working on legacy SQLite builds, but upsert_one and upsert_many will be slower than on modern SQLite.
The pysqlite3 and pysqlite3-binary distributions both provide the same
pysqlite3 import namespace. Do not normally install them together: their
files can overwrite one another, leaving ScriptDB with an old or broken SQLite
build. When this happens, ScriptDB reports the selected backend, SQLite
version, module path, and the detected distribution combination.
Inspect the active SQLite modules and installed distributions with:
python - <<'PY'
import sqlite3
print("stdlib:", sqlite3.sqlite_version, sqlite3.__file__)
try:
import pysqlite3
print("pysqlite3:", pysqlite3.sqlite_version, pysqlite3.__file__)
except ImportError as exc:
print("pysqlite3 import failed:", exc)
PY
python -m pip show pysqlite3
python -m pip show pysqlite3-binary
python -m pip freeze | grep -i sqliteFor a conflicting, stale, or unexpectedly old installation, remove both distributions and install only the binary build:
python -m pip uninstall -y pysqlite3 pysqlite3-binary
python -m pip install \
--no-cache-dir \
--only-binary=:all: \
pysqlite3-binaryThe complete documentation, with examples for the synchronous and
asynchronous APIs, migrations, queries, cache helpers, SQL builders, and
SQLite troubleshooting, is available at
mihanentalpo.github.io/ScriptDB.
The Markdown source is stored in docs/ and is published to GitHub
Pages automatically after changes to the main branch.
To use the synchronous implementation:
pip install scriptdbTo use the asynchronous version (installs aiosqlite):
pip install scriptdb[async]To bundle a modern SQLite build via pysqlite3 for legacy systems:
pip install scriptdb[pysqlite]If you must stay on an older SQLite build, you can also enable the legacy upsert fallback:
from scriptdb import SyncBaseDB
class MyDB(SyncBaseDB):
def migrations(self):
return []
with MyDB.open("app.db", legacy_sqlite_support=True) as db:
db.upsert_one("items", {"id": 1, "value": "x"})This compatibility mode is slower because upserts are emulated without ON CONFLICT ... DO UPDATE.
Both the asynchronous and synchronous interfaces expose the same API.
The only difference is whether methods are coroutines (AsyncBaseDB and
AsyncCacheDB) or regular blocking functions (SyncBaseDB and
SyncCacheDB). Import AsyncBaseDB/AsyncCacheDB from scriptdb.asyncdb for
asynchronous usage or SyncBaseDB/SyncCacheDB from scriptdb.syncdb for
synchronous usage. For convenience, each module also exposes BaseDB and
CacheDB aliases pointing to the respective implementations.
If you need both versions of the same schema, scriptdb.conversion can build a
class that reuses an existing set of migrations without duplicating any code.
Migrations must already match the target style: synchronous migrations for
SyncBaseDB subclasses and asynchronous migrations for AsyncBaseDB
subclasses. Conversion only works for SQL-based migrations (strings or
Builder objects); callable function migrations cannot be converted because
their async/sync behavior is implementation-specific.
from scriptdb.conversion import async_from_sync, sync_from_async
from scriptdb.syncdb import SyncBaseDB
class SyncEventsDB(SyncBaseDB):
def migrations(self):
return [
{
"name": "create_events",
"sql": "CREATE TABLE events(id INTEGER PRIMARY KEY, payload TEXT)",
},
]
# Generate an async counterpart that shares migrations
AsyncEventsDB = async_from_sync(SyncEventsDB)
# Or build a sync wrapper around an async definition
SyncEventsFromAsync = sync_from_async(AsyncEventsDB)Create a subclass of AsyncBaseDB and provide a list of migrations:
from scriptdb import AsyncBaseDB
class MyDB(AsyncBaseDB):
def migrations(self):
return [
{
"name": "create_links",
"sql": """
CREATE TABLE links(
resource_id INTEGER PRIMARY KEY,
referrer_url TEXT,
url TEXT,
status INTEGER,
progress INTEGER,
is_done INTEGER,
content BLOB
)
""",
},
{
"name": "add_created_idx",
# run multiple statements sequentially
"sqls": [
"ALTER TABLE links ADD COLUMN created_at TEXT", # new column
"CREATE INDEX idx_links_created_at ON links(created_at)", # index
],
},
]You can bundle multiple statements into a single migration entry by separating them with semicolons; ScriptDB will
execute them sequentially using SQLite's executescript:
{
"name": "backfill_created_flags",
"sql": """
ALTER TABLE links ADD COLUMN created_flag INTEGER DEFAULT 0;
UPDATE links SET created_flag = 1 WHERE created_at IS NOT NULL;
""",
}Each migration and its applied_migrations marker are committed atomically.
If any statement fails, the whole migration is rolled back.
async def main():
async with MyDB.open("app.db") as db: # WAL journaling is enabled by default
await db.execute(
"INSERT INTO links(url, status, progress, is_done) VALUES(?,?,?,?)",
("https://example.com/data", 0, 0, 0),
)
row = await db.query_one("SELECT url FROM links")
print(row["url"]) # -> https://example.com/data
# Manual open/close without a context manager
db = await MyDB.open("app.db")
try:
await db.execute(
"INSERT INTO links(url, status, progress, is_done) VALUES(?,?,?,?)",
("https://example.com/other", 0, 0, 0),
)
finally:
await db.close()
# Daemonize the aiosqlite worker thread to avoid hanging on exit
async with MyDB.open("app.db", daemonize_thread=True) as db:
await db.execute("SELECT 1")
db = await MyDB.open("app.db", daemonize_thread=True)
try:
await db.execute("SELECT 1")
finally:
await db.close()Always close the database connection with close() or use the async with
context manager as shown above. If you call MyDB.open() without a context
manager, remember to await db.close() when finished. Leaving a database open
may keep background tasks alive and prevent your application from exiting
cleanly.
ScriptDB does not install signal handlers by default, so it will not replace
an application's existing SIGINT or SIGTERM handling. Standalone scripts
can opt in with MyDB.open("app.db", handle_signals=True); the registered
handlers schedule close() on the running event loop.
When enabled, multiple databases share the handler, and ScriptDB restores the
previous application handler after the last database closes.
aiosqlite runs a worker thread to execute SQLite operations. There has been an
ongoing debate in the aiosqlite project about whether this thread should be a
daemon. A non-daemon worker can keep the Python process alive even after all
tasks have finished. ScriptDB ships with the internal
daemonizable_aiosqlite module that wraps aiosqlite.connect and allows this
worker thread to be marked as daemon.
The test suite in this repository relies on this module; without it, lingering
threads would prevent tests from completing. To enable daemon mode in your
application, pass daemonize_thread=True when opening the database as shown
above. Use this option only if your program hangs on exit, as daemon threads can
be terminated abruptly, potentially losing in-flight work.
Create a subclass of SyncBaseDB for blocking use:
from scriptdb import SyncBaseDB
class MyDB(SyncBaseDB):
def migrations(self):
return [
{
"name": "create_links",
"sql": """
CREATE TABLE links(
resource_id INTEGER PRIMARY KEY,
referrer_url TEXT,
url TEXT,
status INTEGER,
progress INTEGER,
is_done INTEGER,
content BLOB
)
""",
},
{
"name": "add_created_idx",
"sqls": [
"ALTER TABLE links ADD COLUMN created_at TEXT", # new column
"CREATE INDEX idx_links_created_at ON links(created_at)", # index
],
},
]
with MyDB.open("app.db") as db: # WAL journaling is enabled by default
db.execute(
"INSERT INTO links(url, status, progress, is_done) VALUES(?,?,?,?)",
("https://example.com/data", 0, 0, 0),
)
row = db.query_one("SELECT url FROM links")
print(row["url"]) # -> https://example.com/data
# Manual open/close without a context manager
db = MyDB.open("app.db")
try:
db.execute(
"INSERT INTO links(url, status, progress, is_done) VALUES(?,?,?,?)",
("https://example.com/other", 0, 0, 0),
)
finally:
db.close()Always close the database connection with close() or use the with
context manager as shown above. Leaving a database open may keep background
tasks alive and prevent your application from exiting cleanly.
The AsyncBaseDB API supports migrations and offers helpers for common operations
and background tasks:
from scriptdb import AsyncBaseDB, run_every_seconds, run_every_queries
class MyDB(AsyncBaseDB):
def migrations(self):
return [
{
"name": "init",
"sql": """
CREATE TABLE links(
resource_id INTEGER PRIMARY KEY,
referrer_url TEXT,
url TEXT,
status INTEGER,
progress INTEGER,
is_done INTEGER,
content BLOB
)
""",
},
{"name": "idx_status", "sql": "CREATE INDEX idx_links_status ON links(status)"},
{"name": "create_meta", "sql": "CREATE TABLE meta(key TEXT PRIMARY KEY, value TEXT)"},
]
# Periodically remove finished links
@run_every_seconds(60)
async def cleanup(self):
await self.execute("DELETE FROM links WHERE is_done = 1")
# Write a checkpoint every 100 executed queries
@run_every_queries(100)
async def checkpoint(self):
await self.execute("PRAGMA wal_checkpoint")
async def main():
async with MyDB.open("app.db") as db: # pass use_wal=False to disable WAL
# Insert many links at once
await db.execute_many(
"INSERT INTO links(url) VALUES(?)",
[("https://a",), ("https://b",), ("https://c",)],
)
# Fetch all URLs
rows = await db.query_many("SELECT url FROM links")
print([r["url"] for r in rows])
# Stream links one by one
async for row in db.query_many_gen("SELECT url FROM links"):
print(row["url"])AsyncBaseDB and SyncBaseDB include convenience helpers for common insert,
update and delete operations:
# Insert one record and get its primary key
pk = await db.insert_one("links", {"url": "https://a"})
# Insert many records
await db.insert_many("links", [{"url": "https://b"}, {"url": "https://c"}])
# Upsert a single record
await db.upsert_one("links", {"resource_id": pk, "status": 200})
# Upsert many records
await db.upsert_many(
"links",
[
{"resource_id": 1, "status": 200},
{"resource_id": 2, "status": 404},
],
)
# Update selected columns in a record
await db.update_one("links", pk, {"progress": 50})
# Delete records
await db.delete_one("links", pk)
await db.delete_many("links", "status = ?", (404,))Group multiple statements into a single unit of work with the built-in transaction
context managers. They automatically call BEGIN, COMMIT, and ROLLBACK for
you:
# Async example
async with db.transaction():
await db.execute("INSERT INTO links(url) VALUES(?)", ("https://example",))
await db.execute("UPDATE links SET status=? WHERE url=?", (200, "https://example"))
# Sync example
with db.transaction():
db.execute("INSERT INTO links(url) VALUES(?)", ("https://example",))
db.execute("UPDATE links SET status=? WHERE url=?", (200, "https://example"))If you need full control, you can also call begin(), commit(), and
rollback() directly on AsyncBaseDB and SyncBaseDB instances. Nested
transactions are not supported—calling begin() or transaction() while a
transaction is already active raises a RuntimeError.
While a transaction is active, operations from other threads or asyncio tasks
wait until it commits or rolls back instead of joining it. Bulk write helpers
also roll back their complete batch if one row fails.
SQLite foreign-key enforcement is enabled for every ScriptDB connection, so
constraints produced by Builder's references= option are enforced.
Query helpers
Scan report · 2026-10-05
- ✓ Prohibited terms or links
- ✓ Repository eligibility
- ✓ slopscore.md paperwork
- ✓ Content policy
- ✓ Risk review
From the balcony · 0 of 2 clapped
Schnitzel and Cap'm Slop read it and passed. Their reasons are on the balcony, with every other verdict.
Critics are accounts on this site with no GitHub account behind them. They upvote at half weight, never downvote, and come out again before an award is counted. Who they are.
0 comments
log in to comment.