SQLite for Operational State in Python Services

Redis is the default choice for job queues and background service state. But Redis is an external dependency — you need to run it, monitor it, and handle reconnection. For single-process Python services, SQLite is a serious alternative: no network, no daemon, ACID transactions, and zero configuration.

The Use Case

In Cognix, three background services run concurrently: the ingestion worker, the intelligence pipeline, and the health check scanner. Each needs to:

Track which files have been processed
Record job status and errors
Store per-service configuration
Report health to a monitoring endpoint

All of this state lives in SQLite.

Schema Design

sql

-- Job tracking
CREATE TABLE jobs (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    job_type    TEXT NOT NULL,          -- 'ingestion', 'intelligence', 'health'
    status      TEXT NOT NULL DEFAULT 'pending',  -- pending | running | done | failed
    source_path TEXT,
    started_at  TIMESTAMP,
    completed_at TIMESTAMP,
    error       TEXT,
    metadata    TEXT                    -- JSON blob for type-specific fields
);
CREATE INDEX idx_jobs_status ON jobs(status);
CREATE INDEX idx_jobs_type ON jobs(job_type, status);
 
-- Service configuration (key-value per service)
CREATE TABLE background_services (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    service_name TEXT NOT NULL UNIQUE,
    enabled     BOOLEAN DEFAULT true,
    interval_seconds INTEGER DEFAULT 300,
    last_run    TIMESTAMP,
    run_count   INTEGER DEFAULT 0,
    config_json TEXT                    -- service-specific config
);
 
-- Health findings
CREATE TABLE health_findings (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    check_type  TEXT NOT NULL,
    severity    TEXT NOT NULL,          -- info | warning | error
    message     TEXT NOT NULL,
    source_path TEXT,
    detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    resolved_at TIMESTAMP,
    is_resolved BOOLEAN DEFAULT false
);

Job Queue Pattern

Enqueue and dequeue without an external broker:

python

import sqlite3
import json
from datetime import datetime
 
def enqueue_job(conn: sqlite3.Connection, job_type: str, source_path: str, metadata: dict = None):
    conn.execute("""
        INSERT INTO jobs (job_type, source_path, metadata)
        VALUES (?, ?, ?)
    """, (job_type, source_path, json.dumps(metadata or {})))
    conn.commit()
 
def dequeue_next_job(conn: sqlite3.Connection, job_type: str) -> dict | None:
    row = conn.execute("""
        SELECT id, job_type, source_path, metadata
        FROM jobs
        WHERE job_type = ? AND status = 'pending'
        ORDER BY id ASC
        LIMIT 1
    """, (job_type,)).fetchone()
 
    if not row:
        return None
 
    # Mark as running atomically
    conn.execute("""
        UPDATE jobs SET status = 'running', started_at = CURRENT_TIMESTAMP
        WHERE id = ?
    """, (row[0],))
    conn.commit()
 
    return {"id": row[0], "type": row[1], "source_path": row[2], "metadata": json.loads(row[3])}
 
def complete_job(conn: sqlite3.Connection, job_id: int):
    conn.execute("""
        UPDATE jobs SET status = 'done', completed_at = CURRENT_TIMESTAMP
        WHERE id = ?
    """, (job_id,))
    conn.commit()
 
def fail_job(conn: sqlite3.Connection, job_id: int, error: str):
    conn.execute("""
        UPDATE jobs SET status = 'failed', completed_at = CURRENT_TIMESTAMP, error = ?
        WHERE id = ?
    """, (error, job_id))
    conn.commit()

Background Worker Loop

python

import threading
import time
 
class IngestionWorker:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._stop = threading.Event()
 
    def run(self):
        conn = sqlite3.connect(self.db_path, check_same_thread=False)
        conn.execute("PRAGMA journal_mode=WAL")  # allow concurrent reads
        conn.execute("PRAGMA synchronous=NORMAL")
 
        while not self._stop.is_set():
            job = dequeue_next_job(conn, "ingestion")
            if job:
                try:
                    self._process(job, conn)
                    complete_job(conn, job["id"])
                except Exception as e:
                    fail_job(conn, job["id"], str(e))
            else:
                self._stop.wait(timeout=5.0)  # poll every 5 seconds
 
    def stop(self):
        self._stop.set()
 
    def _process(self, job: dict, conn: sqlite3.Connection):
        # Parse, chunk, embed, store...
        pass
 
worker = IngestionWorker("./app.db")
thread = threading.Thread(target=worker.run, daemon=True)
thread.start()

WAL Mode

Enable Write-Ahead Logging for concurrent access:

python

conn.execute("PRAGMA journal_mode=WAL")

Without WAL, any write locks the entire database — your API can't read while a background job is writing. WAL allows concurrent reads during writes. Always enable it for background service use.

Health Check State

python

def record_health_finding(conn, check_type: str, severity: str, message: str, source_path: str = None):
    conn.execute("""
        INSERT INTO health_findings (check_type, severity, message, source_path)
        VALUES (?, ?, ?, ?)
    """, (check_type, severity, message, source_path))
    conn.commit()
 
def get_open_findings(conn) -> list[dict]:
    rows = conn.execute("""
        SELECT check_type, severity, message, source_path, detected_at
        FROM health_findings
        WHERE is_resolved = false
        ORDER BY
            CASE severity WHEN 'error' THEN 0 WHEN 'warning' THEN 1 ELSE 2 END,
            detected_at DESC
    """).fetchall()
    return [{"type": r[0], "severity": r[1], "message": r[2], "path": r[3], "at": r[4]}
            for r in rows]

Why Not Redis

Redis advantages: pub/sub, O(1) set operations, sorted sets for priority queues, horizontal scaling.

SQLite advantages for single-process services:

Zero dependency (no external process to manage)
ACID transactions (partial updates never leave the queue in inconsistent state)
SQL queries (filter jobs by type, status, source path with indexes)
Persistent across restarts by default (Redis requires configuration)
Free — no instance cost

For a service running on a single machine handling hundreds to thousands of jobs per hour, SQLite outperforms Redis operationally — there's simply nothing to monitor, restart, or scale.