Skip to content

teff.checkpoint.sqlite

teff.checkpoint.sqlite

SQLite checkpointing — stdlib only, shared file format with the RAG store.

Classes:

Name Description
SQLiteCheckpointer

Store checkpoints in a SQLite database.

SQLiteCheckpointer

Bases: Checkpointer

Store checkpoints in a SQLite database.

Uses one row per (owner, checkpoint_id) pair — a composite primary key, so the same ID can belong to different owners (users/tenants) without colliding. Each save is a single INSERT .. ON CONFLICT REPLACE transaction, so a crash leaves either the old or the new row, never a mix. Existing single-owner databases are migrated in place: their rows move under :data:~teff.checkpoint.DEFAULT_OWNER, and an updated_at column is added for TTL cleanup.

All database work runs in a worker thread (asyncio.to_thread) behind a lock, so checkpoint saves never block the event loop — important when many parallel branches checkpoint through the same store.

Parameters:

Name Type Description Default
path str

Path to the SQLite database file.

 required

Methods:

Name Description
cleanup

Delete stale checkpoints; returns how many were removed.
close

Close the underlying SQLite connection.
list

Return all checkpoint IDs persisted for owner.
Source code in teff/checkpoint/sqlite.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
class SQLiteCheckpointer(Checkpointer):
    """Store checkpoints in a SQLite database.

    Uses one row per ``(owner, checkpoint_id)`` pair — a composite primary
    key, so the same ID can belong to different owners (users/tenants)
    without colliding.  Each ``save`` is a single ``INSERT .. ON CONFLICT
    REPLACE`` transaction, so a crash leaves either the old or the new row,
    never a mix.  Existing single-owner databases are migrated in place:
    their rows move under :data:`~teff.checkpoint.DEFAULT_OWNER`, and an
    ``updated_at`` column is added for TTL cleanup.

    All database work runs in a worker thread (``asyncio.to_thread``) behind
    a lock, so checkpoint saves never block the event loop — important when
    many parallel branches checkpoint through the same store.

    Args:
        path: Path to the SQLite database file.
    """

    def __init__(self, path: str):
        self._path = path
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(path, check_same_thread=False)
        self._migrate()
        self._conn.execute(
            """
            CREATE TABLE IF NOT EXISTS checkpoints (
                owner TEXT NOT NULL DEFAULT 'default',
                checkpoint_id TEXT NOT NULL,
                state TEXT NOT NULL,
                next_node_id TEXT,
                iteration INTEGER NOT NULL,
                updated_at REAL,
                PRIMARY KEY (owner, checkpoint_id)
            )
            """
        )
        self._conn.commit()

    def _migrate(self) -> None:
        """Migrate a legacy single-owner table to the owner-scoped schema."""
        row = self._conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='checkpoints'"
        ).fetchone()
        if row is None:
            return
        cols = [r[1] for r in self._conn.execute("PRAGMA table_info(checkpoints)")]
        if "owner" not in cols:
            self._conn.execute("ALTER TABLE checkpoints RENAME TO checkpoints_legacy")
            self._conn.execute(
                """
                CREATE TABLE checkpoints (
                    owner TEXT NOT NULL DEFAULT 'default',
                    checkpoint_id TEXT NOT NULL,
                    state TEXT NOT NULL,
                    next_node_id TEXT,
                    iteration INTEGER NOT NULL,
                    updated_at REAL,
                    PRIMARY KEY (owner, checkpoint_id)
                )
                """
            )
            self._conn.execute(
                """
                INSERT INTO checkpoints (owner, checkpoint_id, state, next_node_id, iteration)
                SELECT 'default', checkpoint_id, state, next_node_id, iteration
                FROM checkpoints_legacy
                """
            )
            self._conn.execute("DROP TABLE checkpoints_legacy")
            self._conn.commit()
            return
        if "updated_at" not in cols:
            self._conn.execute("ALTER TABLE checkpoints ADD COLUMN updated_at REAL")
            self._conn.commit()

    def close(self) -> None:
        """Close the underlying SQLite connection."""
        with self._lock:
            self._conn.close()

    async def _run(self, fn, *args, **kwargs):
        """Run a sync DB call in a worker thread, serialised by the lock."""

        def _call():
            with self._lock:
                return fn(*args, **kwargs)

        return await asyncio.to_thread(_call)

    async def save(
        self,
        checkpoint_id: str,
        checkpoint: Checkpoint,
        *,
        owner: str = DEFAULT_OWNER,
    ) -> None:
        def _save():
            self._conn.execute(
                """
                INSERT INTO checkpoints (owner, checkpoint_id, state, next_node_id, iteration, updated_at)
                VALUES (?, ?, ?, ?, ?, ?)
                ON CONFLICT(owner, checkpoint_id) DO UPDATE SET
                    state = excluded.state,
                    next_node_id = excluded.next_node_id,
                    iteration = excluded.iteration,
                    updated_at = excluded.updated_at
                """,
                (
                    owner,
                    checkpoint_id,
                    json.dumps(checkpoint.state, ensure_ascii=False),
                    checkpoint.next_node_id,
                    checkpoint.iteration,
                    time.time(),
                ),
            )
            self._conn.commit()

        await self._run(_save)

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        def _load():
            return self._conn.execute(
                "SELECT state, next_node_id, iteration FROM checkpoints "
                "WHERE owner = ? AND checkpoint_id = ?",
                (owner, checkpoint_id),
            ).fetchone()

        row = await self._run(_load)
        if row is None:
            return None
        return Checkpoint(
            state=json.loads(row[0]),
            next_node_id=row[1],
            iteration=row[2],
        )

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        def _delete():
            self._conn.execute(
                "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                (owner, checkpoint_id),
            )
            self._conn.commit()

        await self._run(_delete)

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""

        def _list():
            return self._conn.execute(
                "SELECT checkpoint_id FROM checkpoints WHERE owner = ? ORDER BY checkpoint_id",
                (owner,),
            ).fetchall()

        rows = await self._run(_list)
        return [r[0] for r in rows]

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed."""

        def _cleanup():
            if max_age is None and keep_last is None:
                return 0
            removed = 0
            now = time.time()
            if owner is not None:
                owners = [owner]
            else:
                owners = [
                    r[0]
                    for r in self._conn.execute(
                        "SELECT DISTINCT owner FROM checkpoints"
                    ).fetchall()
                ]
            for own in owners:
                if max_age is not None:
                    cur = self._conn.execute(
                        "DELETE FROM checkpoints WHERE owner = ? AND "
                        "COALESCE(updated_at, 0) < ?",
                        (own, now - max_age),
                    )
                    removed += cur.rowcount
                if keep_last is not None:
                    stale = [
                        r[0]
                        for r in self._conn.execute(
                            "SELECT checkpoint_id FROM checkpoints WHERE owner = ? "
                            "ORDER BY COALESCE(updated_at, 0) DESC LIMIT -1 OFFSET ?",
                            (own, keep_last),
                        ).fetchall()
                    ]
                    for cid in stale:
                        self._conn.execute(
                            "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                            (own, cid),
                        )
                        removed += 1
            self._conn.commit()
            return removed

        return await self._run(_cleanup)

cleanup async 

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

Source code in teff/checkpoint/sqlite.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed."""

    def _cleanup():
        if max_age is None and keep_last is None:
            return 0
        removed = 0
        now = time.time()
        if owner is not None:
            owners = [owner]
        else:
            owners = [
                r[0]
                for r in self._conn.execute(
                    "SELECT DISTINCT owner FROM checkpoints"
                ).fetchall()
            ]
        for own in owners:
            if max_age is not None:
                cur = self._conn.execute(
                    "DELETE FROM checkpoints WHERE owner = ? AND "
                    "COALESCE(updated_at, 0) < ?",
                    (own, now - max_age),
                )
                removed += cur.rowcount
            if keep_last is not None:
                stale = [
                    r[0]
                    for r in self._conn.execute(
                        "SELECT checkpoint_id FROM checkpoints WHERE owner = ? "
                        "ORDER BY COALESCE(updated_at, 0) DESC LIMIT -1 OFFSET ?",
                        (own, keep_last),
                    ).fetchall()
                ]
                for cid in stale:
                    self._conn.execute(
                        "DELETE FROM checkpoints WHERE owner = ? AND checkpoint_id = ?",
                        (own, cid),
                    )
                    removed += 1
        self._conn.commit()
        return removed

    return await self._run(_cleanup)

close 

close()

Close the underlying SQLite connection.

Source code in teff/checkpoint/sqlite.py 
90
91
92
93
def close(self) -> None:
    """Close the underlying SQLite connection."""
    with self._lock:
        self._conn.close()

list async 

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/sqlite.py 
164
165
166
167
168
169
170
171
172
173
174
async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""

    def _list():
        return self._conn.execute(
            "SELECT checkpoint_id FROM checkpoints WHERE owner = ? ORDER BY checkpoint_id",
            (owner,),
        ).fetchall()

    rows = await self._run(_list)
    return [r[0] for r in rows]