16 Keep state in SQLite

SQLite is a file-based database engine that is very useful for storing persistent, structured data like sample identities, protocol start times, and measurements in a 96-well plate. SQLite gives the concurrency and robustness of a database engine while being very easy to spin up and use.

Setup

import json, sqlite3, time
from datetime import datetime
from pathlib import Path

run_id = datetime.now().strftime("%Y%m%d_%H%M%S")

16.1 Track samples and plate maps in SQLite

SQLite is very amenable to storing data about 96-well plates.

A plate map can be represented directly:

db = sqlite3.connect("samples.sqlite")
db.execute("""
    CREATE TABLE IF NOT EXISTS samples (
        run_id      TEXT,
        plate       TEXT,
        well        TEXT,
        sample_id   TEXT,
        volume_ul   REAL,
        recorded    TEXT,
        PRIMARY KEY (run_id, plate, well)
    )
""")

1: A path creates the file if it does not exist; ":memory:" gives a database that disappears with the process.

Insert or update wells:

now = datetime.now().isoformat(timespec="seconds")

db.executemany(
    "INSERT OR REPLACE INTO samples VALUES (?, ?, ?, ?, ?, ?)",
    [
        (run_id, "plate_1", "A1", "S001", 100, now),
        (run_id, "plate_1", "B1", "S002", 100, now),
        (run_id, "plate_1", "C1", "S003",  50, now),
    ],
)
db.commit()

1: Parameter placeholders (?), never string formatting. The primary key is (run_id, plate, well) — one row per well per run — so INSERT OR REPLACE updates a well in place rather than duplicating it.
2: Nothing is written until commit().

Read the plate map later:

rows = db.execute(
    "SELECT well, sample_id, volume_ul FROM samples "
    "WHERE run_id = ? AND plate = ? ORDER BY well",
    (run_id, "plate_1"),
)

for well, sample, volume in rows:
    print(well, sample, volume)

A1 S001 100.0
B1 S002 100.0
C1 S003 50.0

SQLite is useful here because the information is structured and queryable but does not require a database server.

A second table can hold sample history instead of only the latest plate map:

db.execute("""
    CREATE TABLE IF NOT EXISTS sample_events (
        run_id      TEXT,
        sample_id   TEXT,
        event       TEXT,
        value       REAL,
        recorded    TEXT
    )
""")

db.execute(
    "INSERT INTO sample_events VALUES (?, ?, ?, ?, ?)",
    (run_id, "S001", "od", 0.62, now),
)
db.commit()

That lets the same file contain both a current plate map and a simple history of measurements or sample operations.

See also: the one-run formats are in chapter 7; this is the across-runs one.

16.2 Keep the scheduler queue in SQLite

A scheduler becomes more useful when queued work survives the Python process that happens to be executing it. An in-memory asyncio.Queue disappears with its process; SQLite can act as a small persistent job queue instead.

Start with a table:

db.execute("""
    CREATE TABLE IF NOT EXISTS jobs (
        id          INTEGER PRIMARY KEY,
        operation   TEXT NOT NULL,
        resource    TEXT NOT NULL,
        payload     TEXT,
        priority    INTEGER DEFAULT 0,
        status      TEXT DEFAULT 'queued',
        not_before  REAL DEFAULT 0,
        created     REAL NOT NULL,
        started     REAL,
        finished    REAL,
        error       TEXT
    )
""")

Submit work by inserting a row:

def submit(db, operation, resource, payload=None,
           priority=0, not_before=0):
    db.execute(
        """
        INSERT INTO jobs
        (operation, resource, payload, priority, not_before, created)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        (operation, resource, json.dumps(payload or {}),
         priority, not_before, time.time()),
    )
    db.commit()

For example:

submit(db, "read_plate", "reader",
       payload={"plate": "plate_1"}, priority=10)

Find the next eligible job for one resource:

def next_job(db, resource, now):
    return db.execute(
        """
        SELECT id, operation, payload
        FROM jobs
        WHERE status = 'queued'
          AND resource = ?
          AND not_before <= ?
        ORDER BY priority DESC, created ASC
        LIMIT 1
        """,
        (resource, now),
    ).fetchone()

Claim it:

def claim(db, job_id, now):
    cur = db.execute(
        """
        UPDATE jobs
        SET status = 'running',
            started = ?
        WHERE id = ?
          AND status = 'queued'
        """,
        (now, job_id),
    )
    db.commit()

    return cur.rowcount == 1

And finish it. Completing and failing are the same transition — a terminal status, a finish time, and an error if there was one — so they are one function with the status as an argument:

def finish(db, job_id, now, status, error=None):
    db.execute(
        """
        UPDATE jobs
        SET status = ?,
            finished = ?,
            error = ?
        WHERE id = ?
        """,
        (status, now, repr(error) if error is not None else None, job_id),
    )
    db.commit()

The scheduler state is now ordinary persistent data:

queued
running
complete
failed

and work can additionally be ordered by resource, priority, not_before, and created.

The execution runtime is separate. One worker could be synchronous:

#| eval: false
while True:
    now = time.time()
    job = next_job(db, "reader", now)

    if job is None:
        time.sleep(0.1)
        continue

    job_id, operation, payload = job
    if not claim(db, job_id, now):
        continue

    try:
        run(operation, json.loads(payload))
    except Exception as exc:
        finish(db, job_id, time.time(), "failed", exc)
    else:
        finish(db, job_id, time.time(), "complete")

1: Read and claim on the same instant. Another worker may have taken the job in between, which is exactly what claim returning False means.

Another implementation could use threads, processes, asyncio, or workers on separate computers. The queue does not depend on any of them.

16.2.1 Delayed work

not_before is enough to represent a simple incubation:

submit(db, "read_plate", "reader",
       payload={"plate": "plate_1"},
       not_before=time.time() + 1800)

The scheduler does not need to sleep for thirty minutes. The row sits in the table, and next_job passes over it until now >= not_before — no timer, no thread parked on the incubation.

16.3 What to remember

SQLite is the across-process layer: sqlite3 is stdlib, writes one file, and needs no server.
Use ? parameter placeholders and db.commit(); nothing is written until commit.
One row per well per run, keyed on (run_id, plate, well), makes INSERT OR REPLACE an update-in-place.
A persistent job queue is just a table with status, priority, resource, and not_before — the execution runtime (threads, processes, asyncio, remote workers) is a separate concern.