합치기: DB wrap, CRUD method 노출, codebase 의 나머지에서 SQL 숨기는 store class. aiosqlite 사용. 아래는 피파 ConversationStore 의 단순화 버전.
흩어진 query 대신 store class 인 이유:
Schema 가 migration + 테이블 layout 통해 한 자리에 documented.
Route 가 SQL 안 씀, method 호출. SQL 변경 codebase ripple 안 함.
Store 테스트가 unit-test-shaped (in-memory SQLite, 빠름, FastAPI 없음).
Data layer refactoring (예: libSQL 로 swap) = 한 파일 작업.
Self-reference: 피파 backend/store/conversations.py 가 정확히 이 모양 — async class 한 개, connection 한 개, backend/routes/chat.py 의 모든 route 가 SQL embed 안 하고 store method 호출. GET 시 healing 로직도 여기 살아.
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
위 ConversationStore end-to-end 구현. :memory: SQLite 로 create/list/add/messages/delete cover 하는 unit test. 그 다음 a04 lifespan 패턴으로 FastAPI 앱에 wire + HTTP 통한 작은 chat-shaped CRUD test. Brain-specific 필터링이나 페이지네이션 어떻게 확장할지 기록.
Progress
Progress is local-only — sign in to sync across devices.