진짜 ConversationStore — End-to-End CRUD · SQLite Quest

피파가 실제 쓰는 모양

이제 다 합쳐보자. DB를 감싸고, CRUD 메서드를 열어주고, codebase의 나머지에서 SQL을 감추는 store class야. aiosqlite를 쓰고. 아래는 피파의 ConversationStore를 단순하게 줄인 버전이야.

query를 여기저기 흩뿌리지 않고 store class로 묶는 이유가 넷 있어.

schema가 migration과 테이블 정의를 통해 한자리에 문서로 남아.
route가 SQL을 안 쓰고 메서드만 불러. SQL을 고쳐도 codebase 전체로 파문이 안 퍼져.
store 테스트가 단위 테스트 모양이 돼. 메모리 안의 SQLite면 되니까 빠르고 FastAPI도 필요 없어.
데이터 계층을 갈아엎을 때 — 예를 들어 libSQL로 바꿀 때 — 파일 하나만 손보면 돼.

Self-reference: 피파 backend/store/conversations.py가 딱 이 모양이야. async class 하나, connection 하나, 그리고 backend/routes/chat.py의 모든 route가 SQL을 품지 않고 store 메서드만 불러. 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를 처음부터 끝까지 구현해봐. :memory: SQLite로 create, list, add, messages, delete를 전부 덮는 단위 테스트를 쓰고. 그다음 a04의 lifespan 패턴으로 FastAPI 앱에 물려서 HTTP로 도는 작은 chat 모양 CRUD 테스트까지 해봐. brain별 필터링이나 페이지네이션을 어떻게 얹을지도 적어둬.

진짜 ConversationStore — End-to-End CRUD

피파가 실제 쓰는 모양

Code

Exercise

Progress

댓글 0