sqlite-helper-py 1.0.0

Zero-config SQLite helper for Python - migrations, CRUD generation, connection pooling, WAL

sqlite-helper-py

Zero-config SQLite helper for Python — migrations, CRUD generation, connection pooling, and WAL mode in one small, dependency-free package.

Install

pip install sqlite-helper-py

Requires Python 3.9+ and nothing else — the standard library's sqlite3 module does all the work.

---

Quick Start

from sqlite_helper_py import Database, SchemaBuilder, WALConfig

# Open (or create) a database with WAL mode enabled
db = Database("myapp.db", wal_config=WALConfig())

# Define a table with the fluent builder
users_schema = (
    SchemaBuilder("users")
    .integer("id", primary_key=True)
    .text("email", nullable=False, unique=True)
    .text("name", nullable=False)
    .boolean("active", default=True)
    .timestamps()
    .build()
)

# Create the table
db.create_table(users_schema)

# Get a CRUD repository
users = db.repository(users_schema)

# Insert
uid = users.insert({"email": "[e-mail]", "name": "Alice"})

# Fetch by PK
user = users.get(uid)
print(user)  # {'id': 1, 'email': '[e-mail]', 'name': 'Alice', ...}

# Update
users.update(uid, {"name": "Alice Smith"})

# Find by field
results = users.find(active=1)

# Delete
users.delete(uid)

---

Features

Feature	Description
Connection pooling	Thread-safe pool with configurable size and timeout
WAL mode	One-call WAL activation with preset profiles
Schema builder	Fluent DSL for table and index definitions
Migrations	Versioned, checksummed, transactional migrations
CRUD repository	Full Create/Read/Update/Delete with pagination and iteration
Query builder	Parameterised SELECT builder — no raw string concatenation
Zero dependencies	Pure Python, stdlib sqlite3 only

---

Connection Pool

from sqlite_helper_py import ConnectionPool

pool = ConnectionPool("myapp.db", pool_size=10, timeout=30.0)

with pool.connection() as conn:
    conn.execute("INSERT INTO logs (msg) VALUES (?)", ("hello",))

with pool.transaction() as conn:
    conn.execute("UPDATE accounts SET balance = balance - 50 WHERE id = 1")
    conn.execute("UPDATE accounts SET balance = balance + 50 WHERE id = 2")

pool.close()

Context manager

Database implements __enter__ / __exit__:

with Database("myapp.db") as db:
    db.create_table(schema)
    ...
# pool is closed automatically

---

WAL Mode

from sqlite_helper_py import Database, WALConfig, PROFILE_HIGH_CONCURRENCY

# Apply a built-in profile
db = Database("myapp.db", wal_config=PROFILE_HIGH_CONCURRENCY)

# Or fine-tune
cfg = WALConfig(
    synchronous="FULL",        # stronger durability
    wal_autocheckpoint=500,
    cache_size=-65536,         # 64 MB
    busy_timeout=10_000,
)
status = db.enable_wal(cfg)
print(status)

Available profiles: PROFILE_HIGH_CONCURRENCY, PROFILE_HIGH_DURABILITY, PROFILE_EMBEDDED.

Manual checkpoint:

from sqlite_helper_py import checkpoint

with db.connection() as conn:
    wal_frames, checkpointed = checkpoint(conn, mode="TRUNCATE")

---

Schema Builder

from sqlite_helper_py import SchemaBuilder, ForeignKey

posts_schema = (
    SchemaBuilder("posts")
    .integer("id", primary_key=True)
    .text("title", nullable=False)
    .text("body")
    .integer("author_id", foreign_key=ForeignKey("users", on_delete="CASCADE"))
    .integer("view_count", default=0)
    .timestamps()
    .index("idx_posts_author", ["author_id"])
    .index("idx_posts_title", ["title"], unique=True)
    .build()
)

# Print the DDL
print(posts_schema.create_statement())

Direct column definition:

from sqlite_helper_py import TableSchema, Column, ColumnType

schema = TableSchema(
    "events",
    columns=[
        Column("id", ColumnType.INTEGER, primary_key=True, autoincrement=True),
        Column("name", ColumnType.TEXT, nullable=False),
        Column("payload", ColumnType.BLOB),
        Column("ts", ColumnType.TEXT, default="CURRENT_TIMESTAMP"),
    ],
)

---

Migrations

Inline SQL migrations

from sqlite_helper_py import Database, Migration

db = Database("myapp.db")

migrations = [
    Migration(
        version=1,
        name="create_users",
        sql="""
            CREATE TABLE IF NOT EXISTS users (
                id    INTEGER PRIMARY KEY AUTOINCREMENT,
                email TEXT NOT NULL UNIQUE,
                name  TEXT NOT NULL
            );
        """,
    ),
    Migration(
        version=2,
        name="add_users_active",
        sql="ALTER TABLE users ADD COLUMN active INTEGER NOT NULL DEFAULT 1;",
    ),
]

applied = db.migrate(migrations)
print(f"Applied: {applied}")  # [1, 2]

Python function migrations

import sqlite3
from sqlite_helper_py import Migration

def seed_data(conn: sqlite3.Connection) -> None:
    conn.execute("INSERT INTO config (key, value) VALUES ('version', '1.0')")

def remove_seed(conn: sqlite3.Connection) -> None:
    conn.execute("DELETE FROM config WHERE key = 'version'")

Migration(
    version=3,
    name="seed_config",
    up_fn=seed_data,
    down_fn=remove_seed,
)

Migration status and rollback

runner = db.migration_runner()
runner.register(*migrations)

for row in runner.status():
    state = "✓" if row["applied"] else "○"
    print(f"  {state} [{row['version']}] {row['name']}")

# Roll back the latest migration
runner.rollback(2)

SQL file migrations

Name files {version}_{name}.sql and point the runner at the directory:

migrations/
    0001_create_users.sql
    0002_add_index.sql
    0003_add_posts.sql

runner = db.migration_runner()
runner.register_sql_directory("migrations/")
runner.migrate()

---

CRUD Repository

users = db.repository(users_schema)

# Insert / upsert
uid  = users.insert({"email": "[e-mail]", "name": "Bob"})
uid2 = users.upsert({"email": "[e-mail]", "name": "Bob Updated"}, ["email"])

# Fetch
user = users.get(uid)                   # by PK, returns dict or None
user = users.get_or_raise(uid)          # raises RecordNotFoundError on miss
user = users.find_one(email="[e-mail]")

# List
all_users  = users.all(order_by="name ASC")
active     = users.find(active=1)
count      = users.count(active=1)
exists     = users.exists(email="[e-mail]")

# Pagination
page = users.paginate(page=1, page_size=20, order_by="name ASC")
# {'items': [...], 'page': 1, 'page_size': 20, 'total': 150, 'total_pages': 8}

# Memory-efficient iteration
for user in users.iter_all(batch_size=500):
    process(user)

# Update
users.update(uid, {"name": "Robert"})
users.update_where({"active": 0}, name="Deleted")

# Delete
users.delete(uid)
users.delete_where(active=0)

---

Query Builder

from sqlite_helper_py import QueryBuilder

sql, params = (
    QueryBuilder("orders")
    .select("id", "total", "status")
    .where("status = ?", "pending")
    .where("total > ?", 100)
    .where_in("user_id", [1, 2, 3])
    .order_by("created_at DESC")
    .limit(50)
    .offset(100)
    .build()
)

with db.connection() as conn:
    rows = conn.execute(sql, params).fetchall()

---

API Reference

Database

Method	Description
Database(path, pool_size, wal_config, pragmas)	Open/create database
.connection()	Context manager yielding a pooled connection
.transaction()	Context manager with auto-commit / rollback
.enable_wal(config)	Activate WAL mode, returns WALStatus
.create_table(schema)	Execute CREATE TABLE + indexes
.drop_table(schema)	Execute DROP TABLE
.table_exists(name)	Return bool
.list_tables()	Return list of table names
.repository(schema)	Return cached Repository
.migrate(migrations)	Register + apply migrations
.migration_runner()	Return bound MigrationRunner
.execute(sql, params)	Raw execute
.fetchall(sql, params)	Return list of dicts
.fetchone(sql, params)	Return dict or None
.scalar(sql, params)	Return first column of first row
.integrity_check()	Return True if database is intact
.database_size_bytes()	Approximate size in bytes
.close()	Release all pooled connections

Repository

Method	Description
.insert(data)	Insert row, return PK
.insert_many(rows)	Bulk insert, return count
.upsert(data, conflict_columns)	Insert or update
.get(pk)	Fetch by PK or None
.get_or_raise(pk)	Fetch by PK or raise RecordNotFoundError
.find(**criteria)	List matching rows
.find_one(**criteria)	First matching row or None
.all(order_by, limit, offset)	All rows
.count(**criteria)	Row count
.exists(**criteria)	True if any matching row
.paginate(page, page_size)	Paginated result dict
.iter_all(batch_size)	Iterator over all rows
.update(pk, data)	Update by PK
.update_where(data, **criteria)	Bulk update
.delete(pk)	Delete by PK
.delete_where(**criteria)	Bulk delete
.truncate()	Delete all rows
.query()	Return a QueryBuilder for this table

---

License

MIT — see LICENSE.

---

Vladyslav Zaiets — sarmkadan.com