How I Use Python's sqlite3 Module to Keep Small Script Data Organized

Many of my scripts start by writing results to a CSV file. That works until I need to answer a question like "which runs failed this week" or stop a script from processing the same record twice. At that point I reach for SQLite. It ships with Python as the sqlite3 module, it keeps everything in one file, and there is no server to install or keep running.

Here is the pattern I use for small tools.

Opening a database

import sqlite3
from pathlib import Path

DB_PATH = Path(__file__).resolve().parent / "jobs.db"

def connect():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn

connect creates the file if it does not exist. Setting row_factory to sqlite3.Row lets me read columns by name, like row["status"], instead of remembering positions. Foreign key checks are off by default in SQLite, so I turn them on for every connection.

Creating tables that can run twice

from contextlib import closing

SCHEMA = """
CREATE TABLE IF NOT EXISTS runs (
    id INTEGER PRIMARY KEY,
    source TEXT NOT NULL,
    started_at TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('ok', 'failed')),
    rows_processed INTEGER NOT NULL DEFAULT 0
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_runs_source_start
    ON runs (source, started_at);
"""

with closing(connect()) as conn:
    conn.executescript(SCHEMA)

IF NOT EXISTS means the setup can run on every start without errors. The CHECK constraint stops a typo like "sucess" from ever being stored. The unique index keeps duplicates out: the same source cannot be recorded twice for the same start time.

Inserting rows with placeholders

The most important rule: never build SQL by pasting values into a string. Use placeholders and let the driver handle quoting.

from datetime import datetime, timezone

def record_run(conn, source, status, rows):
    started = datetime.now(timezone.utc).isoformat(timespec="seconds")
    conn.execute(
        "INSERT OR IGNORE INTO runs (source, started_at, status, rows_processed) "
        "VALUES (?, ?, ?, ?)",
        (source, started, status, rows),
    )

The question marks are filled in safely, so a value containing a quote mark cannot break the statement or change what it does. INSERT OR IGNORE works together with the unique index: a duplicate is skipped quietly instead of raising an error. When I want to know about duplicates, I use a plain INSERT and catch sqlite3.IntegrityError instead.

For many rows at once, executemany takes a list of tuples and is much faster than a loop of single inserts.

Transactions and closing the connection

Using a connection in a with block commits when the block finishes and rolls back if an exception is raised. What it does not do is close the connection. That one bit me: I assumed it did, and ended up with a locked database file on Windows. Now I nest two blocks.

from contextlib import closing

with closing(connect()) as conn:
    with conn:
        record_run(conn, "daily_export", "ok", 240)

The outer block closes the file handle. The inner block wraps the work in a transaction, so if something fails halfway through, nothing partial is saved.

Querying

def failed_runs_since(conn, since_iso):
    cur = conn.execute(
        "SELECT source, started_at, rows_processed FROM runs "
        "WHERE status = 'failed' AND started_at >= ? "
        "ORDER BY started_at DESC",
        (since_iso,),
    )
    return [dict(row) for row in cur]

ISO 8601 timestamps stored as text sort correctly as plain strings, so date comparisons work without a separate date type, as long as every row uses the same format and time zone. Converting each Row to a dict makes the result easy to print or dump to JSON.

A quick summary is one query:

summary = conn.execute(
    "SELECT status, COUNT(*) AS runs, SUM(rows_processed) AS total "
    "FROM runs GROUP BY status"
).fetchall()
for row in summary:
    print(row["status"], row["runs"], row["total"])

Looking at the data by hand

When I want to check what a script actually wrote, I open the file in the sqlite3 shell:

sqlite3 jobs.db
.tables
.schema runs
SELECT * FROM runs ORDER BY started_at DESC LIMIT 5;

If the shell is not installed, DB Browser for SQLite gives the same view with a simple interface.

Backups in three lines

with closing(connect()) as src, closing(sqlite3.connect("jobs_backup.db")) as dst:
    src.backup(dst)

backup copies a consistent snapshot even if another connection is open, which is safer than copying the file while a script might be writing to it.

When SQLite is the wrong tool

SQLite is built for one machine. If several services need to write at the same time over a network, use a database server like PostgreSQL. I also keep the database file out of folders that a cloud drive syncs while a script is writing, because a sync in the middle of a write can leave a damaged copy.

What I took away

In delivery and logistics operations, the dispatch board was the single place that showed what went out, what was late and what was still waiting. A small SQLite table does the same job for my scripts. If you have a script that appends to a CSV and you search through it later to find problems, try moving that one log into SQLite. It is usually an hour of work, and the questions you could not easily answer become one query.

Originally published at tonycantave.com.