Look Ma, no ORM !

Update: this is now available as a python module.

Say you want to use SQLite for your next small app and want to avoid pulling a big ORM... but you still want to add behaviour to your data and you still want to have some minimal management of your database migrations.

#!/usr/bin/env -S uv run --script

# /// script
# dependencies = ["Faker"]
# [tool.uv]
# exclude-newer = "2025-03-25T00:00:00Z"
# ///
import sqlite3
from typing import Union, List, Iterable, Any
from pathlib import Path
from faker import Faker
from dataclasses import dataclass, asdict, is_dataclass


class DB:
    def __init__(
        self,
        db_path: Union[str, Path],
        migrations: dict = None,
        migrate_on_connect=False,
        server_mode: bool = True,
    ):
        self.migrate_on_connect = migrate_on_connect
        self.db_path = db_path
        self.server_mode = server_mode
        if migrations:
            self.migrations = migrations
        else:
            self.migrations = {}
        self._connection = None

    @property
    def connection(self) -> sqlite3.Connection:
        if self._connection is None:
            self._connection = sqlite3.connect(self.db_path, check_same_thread=False)
            # data structure for rows that supports casting to dict
            self._connection.row_factory = sqlite3.Row
            cursor = self._connection.cursor()
            cursor.execute("PRAGMA foreign_keys=ON;")
            cursor.execute('PRAGMA encoding = "UTF-8";')
            if self.migrate_on_connect:
                self.migrate()
                self.commit()
            # this is a set of configuration directives
            # to make SQLite a little more fit for
            # concurrent usage (see https://kerkour.com/sqlite-for-servers)
            if self.server_mode:
                cursor.execute("PRAGMA journal_mode = WAL;")
                cursor.execute("PRAGMA busy_timeout = 5000;")
                cursor.execute("PRAGMA synchronous = NORMAL;")
                cursor.execute("PRAGMA cache_size = 1000000000;")
                cursor.execute("PRAGMA temp_store = memory;")
        return self._connection

    def migrate(self, migrations: dict = None):
        if migrations:
            self.migrations = migrations
        cursor = self.cursor()
        cursor.execute( # mandatory table to keep track of migrations
            "CREATE TABLE IF NOT EXISTS _migrations(identifier TEXT UNIQUE, script TEXT);"
        )
        needs_commit = False
        for identifier, script in self.migrations.items():
            cursor.execute(
                "SELECT COUNT(*) from _migrations where identifier = ?", (identifier,)
            )
            result = cursor.fetchone()
            if result["COUNT(*)"] == 0:
                cursor.executescript(script)
                cursor.execute(
                    "INSERT INTO _migrations(identifier, script) VALUES(?,?)",
                    (identifier, script),
                )
                needs_commit = True
        if needs_commit:
            self.commit()
        return self

    def close(self):
        if self._connection is not None:
            self._connection.close()
            self._connection = None

    def insert(self, table: str, values: dict) -> sqlite3.Row:
        if len(values) == 0:
            return {}
        sql = f"INSERT INTO {table}({', '.join(values.keys())}) VALUES({', '.join(['?' for x in values.keys()])}) RETURNING *;"
        cur = self.cursor()
        cur.execute(sql, list(values.values()))
        return cur.fetchone()

    def update(
        self,
        table: str,
        values: Union[dict, dataclass],
        where: str = None,
        params: Any = None,
    ) -> sqlite3.Cursor:
        if is_dataclass(values):
            values = asdict(values)
        if params is not None:
            if type(params) not in [tuple, list]:
                params = [params]
        sql = f"UPDATE {table} SET {', '.join([f'{x}=?' for x in values.keys()])}{f' WHERE {where}' if where is not None else ''} RETURNING *;"
        all_params = list(values.values()) + (params if params is not None else [])
        cur = self.cursor()
        cur.execute(sql, all_params)
        return cur

    def delete(self, table, where: str = None, params: List = None) -> sqlite3.Cursor:
        if params is not None:
            if type(params) not in [tuple, list]:
                params = [params]        
        sql = f"DELETE fROM {table}{f' WHERE {where}' if where is not None else ''} RETURNING *;"
        cur = self.cursor()
        cur.execute(sql, params if params is not None else [])
        return cur

    def execute(self, sql: str, parameters: Iterable[Any] = None) -> sqlite3.Cursor:
        if parameters is not None:
            return self.cursor().execute(sql, parameters)
        return self.cursor().execute(sql)

    def commit(self):
        self.connection.commit()

    def cursor(self) -> sqlite3.Cursor:
        return self.connection.cursor()

    def __del__(self):
        self.close()


if __name__ == "__main__":

    # Testing now...

    db = DB(
        db_path=":memory:",
        migrations={
            # here's the initial migration:
            "0000": "CREATE TABLE IF NOT EXISTS users(id INTEGER PRIMARY KEY, name TEXT UNIQUE NOT NULL);",
            # simulating a schema evolution, let's add a field:
            "0001": "ALTER TABLE users ADD COLUMN email TEXT NULL;",
        },
    ).migrate()

    fake = Faker()
    for i in range(10):
        db.insert("users", {"name": fake.unique.name()})


    # dataclasses are perfect to represent rows
    # while still allowing custom behaviour
    @dataclass
    class User:
        id: int
        name: str
        email: str

        def __str__(self):
            return self.name


    # fetch a random instance
    user = User(
        **dict(db.execute("select * from users ORDER BY RANDOM() limit 1").fetchone())
    )
    print(f"User(id {user.id}), original name: {user}")

    # change name and email
    user.name = fake.unique.name()
    user.email = fake.unique.email()

    # update the instance and get it back from database
    user = User(**dict(db.update("users", user, "id=?", user.id).fetchone()))
    print(f"User(id {user.id}), updated name: {user} <{user.email}>")