Basic SQLite helper for Python: connect with sensible defaults, execute/query helpers, and dict-like rows (no deps).

python-sqlite-helper.py

#!/usr/bin/env python3
"""
Basic SQLite helper (no dependencies).

Features:
- Context-managed connection
- Row results as dict-like objects (sqlite3.Row)
- Helpers for execute / executemany / fetchone / fetchall
- Optional pragmas for better defaults

Works with Python's built-in sqlite3 module.
"""

from __future__ import annotations

import sqlite3
from contextlib import contextmanager
from typing import Any, Iterable, Iterator, Mapping, Optional, Sequence


def connect_db(
    path: str,
    *,
    pragmas: Optional[Mapping[str, Any]] = None,
    timeout: float = 5.0,
) -> sqlite3.Connection:
    """
    Create a SQLite connection with sensible defaults.

    Args:
      path: DB file path (use ':memory:' for in-memory)
      pragmas: optional PRAGMA settings
      timeout: busy timeout in seconds

    Returns:
      sqlite3.Connection
    """
    conn = sqlite3.connect(path, timeout=timeout)
    conn.row_factory = sqlite3.Row  # dict-like rows
    conn.execute("PRAGMA foreign_keys = ON;")

    if pragmas:
        for key, value in pragmas.items():
            conn.execute(f"PRAGMA {key} = {value};")

    return conn


@contextmanager
def db_session(path: str, *, pragmas: Optional[Mapping[str, Any]] = None) -> Iterator[sqlite3.Connection]:
    """
    Context manager that commits on success and rolls back on error.
    """
    conn = connect_db(path, pragmas=pragmas)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def execute(conn: sqlite3.Connection, sql: str, params: Sequence[Any] = ()) -> int:
    """
    Execute a statement and return affected rowcount.
    """
    cur = conn.execute(sql, params)
    return cur.rowcount


def executemany(conn: sqlite3.Connection, sql: str, rows: Iterable[Sequence[Any]]) -> int:
    """
    Execute the same statement for multiple rows and return total affected count.
    """
    cur = conn.executemany(sql, rows)
    return cur.rowcount


def fetchone(conn: sqlite3.Connection, sql: str, params: Sequence[Any] = ()) -> Optional[dict]:
    """
    Run a query and return one row as a dict, or None.
    """
    cur = conn.execute(sql, params)
    row = cur.fetchone()
    return dict(row) if row is not None else None


def fetchall(conn: sqlite3.Connection, sql: str, params: Sequence[Any] = ()) -> list[dict]:
    """
    Run a query and return all rows as list[dict].
    """
    cur = conn.execute(sql, params)
    return [dict(r) for r in cur.fetchall()]


# ---------------------------------------------------------------------------
# Example usage
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    db_path = "example.db"

    pragmas = {
        # Good general-purpose defaults for many apps:
        # "journal_mode": "'WAL'",     # uncomment for better concurrency
        # "synchronous": "NORMAL",     # pairs well with WAL
        "busy_timeout": 5000,          # ms
    }

    with db_session(db_path, pragmas=pragmas) as conn:
        execute(
            conn,
            """
            CREATE TABLE IF NOT EXISTS users (
              id INTEGER PRIMARY KEY AUTOINCREMENT,
              name TEXT NOT NULL,
              email TEXT UNIQUE NOT NULL
            );
            """,
        )

        execute(conn, "INSERT OR IGNORE INTO users (name, email) VALUES (?, ?);", ("Brandon", "brandon@example.com"))

        user = fetchone(conn, "SELECT id, name, email FROM users WHERE email = ?;", ("brandon@example.com",))
        print("one:", user)

        users = fetchall(conn, "SELECT id, name, email FROM users ORDER BY id DESC;")
        print("all:", users)