`without_durability_sqlite`¶

A without-durability checkpoint store and queue backed by one SQLite file, with no server and no third-party driver.

without_durability_sqlite ¶

SCHEMA `module-attribute` ¶

SCHEMA = "\nCREATE TABLE IF NOT EXISTS workflow_checkpoint (\n    workflow TEXT NOT NULL,\n    step TEXT NOT NULL,\n    value TEXT NOT NULL,\n    PRIMARY KEY (workflow, step)\n) WITHOUT ROWID;\n\nCREATE TABLE IF NOT EXISTS workflow_claim (\n    workflow TEXT PRIMARY KEY,\n    token INTEGER NOT NULL,\n    held_until REAL NOT NULL\n) WITHOUT ROWID;\n\nCREATE TABLE IF NOT EXISTS workflow_queue (\n    namespace TEXT NOT NULL,\n    workflow TEXT NOT NULL,\n    visible_at REAL NOT NULL,\n    PRIMARY KEY (namespace, workflow)\n) WITHOUT ROWID;\n\nCREATE INDEX IF NOT EXISTS workflow_queue_visible_at ON workflow_queue (namespace, visible_at);\n"

SqliteEffect ¶

SqliteEffect = Callable[[Cursor], object]

Database `dataclass` ¶

Database(connection: Connection, guard: Lock = Lock())

One SQLite connection and the lock that keeps one caller in it at a time.

The analogue of the Postgres store's connection pool, and the opposite shape for the opposite reason: a pool exists so several statements run at once, and this exists so they do not. Not because a connection would corrupt (SQLite is built serialized here, so it would not), but because a transaction belongs to the connection: without this, a caller arriving mid-BEGIN IMMEDIATE writes into somebody else's transaction and loses its write to that transaction's rollback. See the note at the top of this module for why the event loop's single thread does not already prevent that.

Build it with connect, which applies the pragmas that make this durable rather than merely persistent. Share one between the checkpoint store and the queue: that is what makes SqliteDurable.arrive a single commit, and it is checked rather than assumed.

connection `instance-attribute` ¶

connection: Connection

guard `class-attribute` `instance-attribute` ¶

guard: Lock = field(
    default_factory=asyncio.Lock, repr=False, compare=False
)

run `async` ¶

run(work: Callable[[Connection], T]) -> T

Do work against the connection, on a thread, with nobody else inside it.

Cancellation is where "nobody else" has to be arranged rather than assumed, and it is the reason this is not simply async with self.guard. A thread is not cancellable: cancelling the caller unwinds this coroutine at once while the thread runs on, so releasing the guard on the way out would hand the connection to the next caller while the last one is still inside it. That is not a theoretical race. The statement in flight may be a BEGIN IMMEDIATE transaction, and a write that lands in somebody else's open transaction is committed or rolled back with it: record returns, a read sees the row, and the rollback takes it away again, which is precisely the guarantee this store exists to make.

So the guard is released by the thread finishing rather than by this coroutine returning. The work is a task, the caller awaits a shield of it (so cancelling the caller leaves the task alone), and a done-callback lets go of the connection when it is genuinely free. A cancelled caller still unwinds immediately; what it no longer does is take the connection with it.

What the shield adds beyond that is the reporting. A statement that fails after its caller has gone has nobody left to raise to, and shield hands it to the loop's exception handler rather than dropping it, so a write that failed on the way out of a process is in the log instead of nowhere.

SqliteCheckpointer `dataclass` ¶

SqliteCheckpointer(
    database: Database, codec: CheckpointCodec[str] = JSON
)

A workflow's completed steps as rows in one file, and its claim as a row beside them.

The Checkpointer implementation for a deployment that is one machine, and the one that needs nothing installed. It meets the same requirements as the others by the simplest route any of them take: SQLite admits one writer, so a single statement or a single BEGIN IMMEDIATE transaction is already all the exclusion this needs.

SqliteEffect is a callback over the open transaction's cursor, so a step whose effect is a write to this file happens exactly once. Since the file is the whole datastore, that covers every table an application on this machine keeps here, which is a broader reach than it sounds: it is the same guarantee DBOS gets from Postgres, for an application that never needed Postgres.

A workflow id carries no constraints at all: it is bound as a query parameter, never parsed as key structure. Nothing here derives one id from another either, so an application is free to name a workflow's sibling (a saga's rollback, say) however it likes out of its own namespace.

codec is how a step's result becomes the TEXT in a row and comes back, defaulting to the stdlib's JSON. Swap it to widen what a step may return or to speed the encoding up; what it MUST keep is the round trip, since a resumed pass reads what it produced.

database `instance-attribute` ¶

database: Database

codec `class-attribute` `instance-attribute` ¶

codec: CheckpointCodec[str] = JSON

load `async` ¶

load(workflow: str) -> dict[str, object]

claim `async` ¶

claim(workflow: str, lease: timedelta) -> Pass | None

record `async` ¶

record(holder: Pass, key: str, value: object) -> Recorded

transact `async` ¶

transact(
    holder: Pass, key: str, effect: SqliteEffect
) -> object

Run effect and record it in one transaction, so the step happens once.

The order is the other stores': fence first, because a superseded pass must not act; then the existence check, because a step already recorded must not run again, which is what makes a replay perform nothing at all; then the effect; then the record. BEGIN IMMEDIATE holds the write lock across all four, so no other writer can land between them and any exception rolls back the effect along with its record.

The effect's result is written and read back through the codec rather than returned as it came, so it round-trips exactly as a later pass will see it.

supply `async` ¶

supply(workflow: str, key: str, value: object) -> object

release `async` ¶

release(holder: Pass) -> None

SqliteDurable `dataclass` ¶

SqliteDurable(
    checkpointer: SqliteCheckpointer,
    scheduler: SqliteScheduler,
)

A Durable whose two stores are one file, so arrive is a single commit.

The strongest form of the guarantee, reached by the least machinery: there is nothing to co-locate, no pool to share by accident, and no sharding to grow into. The two stores MUST hold the same Database, checked at construction, which here is less a warning about distributed transactions than a way of saying that two SQLite files are two datastores however adjacent they sit on disk.

checkpointer `instance-attribute` ¶

checkpointer: SqliteCheckpointer

scheduler `instance-attribute` ¶

scheduler: SqliteScheduler

arrive `async` ¶

arrive(workflow: str, key: str, value: object) -> object

Record the value and make the workflow ready, together or not at all.

SqliteScheduler `dataclass` ¶

SqliteScheduler(
    database: Database,
    namespace: str = "workflow",
    lease: timedelta = LEASE,
    poll: timedelta = POLL,
    now: Callable[[], datetime] = now_utc,
)

Scheduler as one table, each row scored by when its workflow becomes visible.

A drop-in for every other queue here, and modelled on the same visibility scheme: queued now is a visible_at in the past, sleeping is one in the future, and being worked on is one a lease ahead, so wake_due, reclaim, and prepare's queue half all have nothing to do.

It polls, like the other visibility-scored queues, so the poll interval is a floor under how fast anything starts. SQLite offers no blocking read and no notification a process outside this one can wait on, so unlike the Postgres store there is not even a LISTEN/NOTIFY left on the table: within one process an asyncio.Event would do it, across processes on one machine it would take a filesystem watch, and neither is here.

database `instance-attribute` ¶

database: Database

namespace `class-attribute` `instance-attribute` ¶

namespace: str = 'workflow'

lease `class-attribute` `instance-attribute` ¶

lease: timedelta = LEASE

poll `class-attribute` `instance-attribute` ¶

poll: timedelta = POLL

now `class-attribute` `instance-attribute` ¶

now: Callable[[], datetime] = now_utc

lease_seconds `class-attribute` `instance-attribute` ¶

lease_seconds: float = field(
    init=False, repr=False, compare=False
)

poll_seconds `class-attribute` `instance-attribute` ¶

poll_seconds: float = field(
    init=False, repr=False, compare=False
)

prepare `async` ¶

prepare() -> None

Create the tables, which every worker does at boot and all but the first find done.

make_ready `async` ¶

make_ready(workflow: str) -> None

wake_at `async` ¶

wake_at(delivery: Delivery, when: datetime) -> None

Suspend the workflow until when, unless something asked for a pass meanwhile.

The receipt is the visibility this pass took, so anything that rescheduled the workflow since (a confirmation, another worker taking over an overrun) wrote a different one and this leaves it be. Which is the right answer rather than a concession: the deadline lives in the workflow's checkpoint, so the pass that runs sooner reaches the same sleep and writes it again.

schedule `async` ¶

schedule(workflow: str, visible_at: datetime) -> None

wake_due `async` ¶

wake_due(now: datetime) -> tuple[str, ...]

Nothing to do: a workflow whose visible_at has passed is already visible.

next_ready `async` ¶

next_ready(within: timedelta) -> Delivery | None

The next visible workflow, waiting up to within for one to appear.

reclaim `async` ¶

reclaim(idle: timedelta) -> Delivery | None

Nothing to take over by hand: an abandoned workflow becomes visible on its own.

done `async` ¶

done(delivery: Delivery) -> None

Drop the workflow, unless something asked for another pass while this one ran.

The receipt is the visibility this pass took, so anything that rescheduled the workflow meanwhile (a confirmation, this pass's own wake_at, another worker taking over an overrun) wrote a different one and this leaves it alone.

connect ¶

connect(
    path: Path | str,
    *,
    timeout: timedelta = timedelta(seconds=5),
) -> Database

Open the database this store runs on, configured for durability rather than speed.

journal_mode=WAL so a reader does not block the writer, which is what lets a status query run while a pass is mid-transaction.
synchronous=FULL because this store's entire claim is that record returning means the value survives. NORMAL is the usual advice under WAL and it trades exactly that away: a commit can be lost on power loss or an OS crash. Everything run_durably reasons about assumes the commit held, so this pays the fsync.
busy_timeout so a second process finding the write lock taken waits for it rather than failing immediately, which is the ordinary case when two processes share the file.

autocommit=True leaves transaction control here rather than in the driver: every statement below is either atomic on its own or wrapped in an explicit BEGIN IMMEDIATE, and nothing is left to a hidden implicit transaction.

migrate `async` ¶

migrate(database: Database) -> None

Create the three tables, from every process, as often as it likes.

No advisory lock and no race to guard against, unlike the Postgres migration: SQLite runs the whole script in one exclusive transaction, so a second process either waits for it or finds the tables already there.

Schema migration as a whole is not what this is. There is no versioning and no path from one shape of these tables to another; user_version is where SQLite keeps that, and a deployment that needs it should use it.

without_durability_sqlite¶