Putting it together: a store class that wraps the database, exposes CRUD methods, hides SQL from the rest of the codebase, and uses aiosqlite throughout. The pattern below is a stripped-down version of Pippa's ConversationStore.
Why a store class over scattered queries:
Schema is documented in one place via the migrations + table layout.
Routes don't write SQL; they call methods. SQL changes don't ripple through the codebase.
Testing the store is unit-test-shaped (in-memory SQLite, fast, no FastAPI).
Refactoring the data layer (e.g., swapping to libSQL) is one file's job.
Self-reference: Pippa's backend/store/conversations.py follows this exact shape — one async class, one connection, every route in backend/routes/chat.py calls store methods rather than embedding SQL. The healing-on-GET logic also lives here, not in routes.
Code
ConversationStore — schema + CRUD·python
import aiosqlite
from typing import Iterable
SCHEMA = '''
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
brain TEXT NOT NULL DEFAULT 'claude',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
conversation_id INTEGER NOT NULL
REFERENCES conversations(id) ON DELETE CASCADE,
role TEXT NOT NULL CHECK (role IN ('user','assistant','system')),
content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_msg_conv_created
ON messages(conversation_id, created_at DESC);
'''
class ConversationStore:
def __init__(self, conn: aiosqlite.Connection):
self.conn = conn
@classmethod
async def open(cls, path: str) -> 'ConversationStore':
conn = await aiosqlite.connect(path)
conn.row_factory = aiosqlite.Row
await conn.execute('PRAGMA journal_mode = WAL')
await conn.execute('PRAGMA foreign_keys = ON')
await conn.execute('PRAGMA busy_timeout = 5000')
await conn.executescript(SCHEMA)
await conn.commit()
return cls(conn)
async def create_conversation(self, title: str, brain: str = 'claude') -> int:
row = await (await self.conn.execute(
'INSERT INTO conversations(title, brain) VALUES (?, ?) RETURNING id',
(title, brain),
)).fetchone()
await self.conn.commit()
return row['id']
async def add_message(self, conv_id: int, role: str, content: str) -> dict:
row = await (await self.conn.execute(
'INSERT INTO messages(conversation_id, role, content) '
'VALUES (?, ?, ?) RETURNING id, created_at',
(conv_id, role, content),
)).fetchone()
await self.conn.execute(
'UPDATE conversations SET updated_at = datetime(\'now\') WHERE id = ?',
(conv_id,),
)
await self.conn.commit()
return dict(row)
async def list_conversations(self, limit: int = 50) -> list[dict]:
async with self.conn.execute(
'SELECT id, title, brain, created_at, updated_at '
'FROM conversations ORDER BY updated_at DESC LIMIT ?',
(limit,),
) as cur:
return [dict(r) async for r in cur]
async def messages_for(self, conv_id: int) -> list[dict]:
async with self.conn.execute(
'SELECT id, role, content, created_at FROM messages '
'WHERE conversation_id = ? ORDER BY created_at',
(conv_id,),
) as cur:
return [dict(r) async for r in cur]
async def delete_conversation(self, conv_id: int) -> int:
cur = await self.conn.execute(
'DELETE FROM conversations WHERE id = ?', (conv_id,)
)
await self.conn.commit()
return cur.rowcount
Exercise
Implement the ConversationStore above end-to-end. Add unit tests using :memory: SQLite that cover create/list/add/messages/delete. Then wire it into a FastAPI app with the lifespan pattern from a04 and run a small chat-shaped CRUD test through HTTP. Note where you'd extend it to add brain-specific filtering or pagination.
Progress
Progress is local-only — sign in to sync across devices.