Schema 마이그레이션과 버저닝

Schema 를 코드처럼 다뤄

Production SQLite schema 는 시간에 따라 진화. 미리 답해야 할 두 질문:

Fresh 설치가 DB 를 현재 버전까지 어떻게 끌어오나?
기존 설치가 N -> N+1 어떻게 안전하게 업그레이드?

SQLite 가 PRAGMA user_version 줘 — DB 헤더에 저장된 32-bit 정수. 마이그레이션 cursor 로 사용. 패턴 (피파의 backend/store/conversations.py + 수많은 다른 곳에서 사용):

시작 시 PRAGMA user_version 읽기.
현재에서 target 까지 각 마이그레이션 step 을 transaction 안에서 적용.
각 마이그레이션 마지막 statement 로 user_version 증가.

Self-reference: 피파의 SQLite store 가 정확히 이 패턴 — backend/store/conversations.py 의 _init_db() 봐. 컬럼/테이블 추가할 때마다 user_version 증가. Conversation GET 마다 healing 로직이 schema 와 JSONL ground truth 정렬 유지.

Code

마이그레이션 runner — 순수 SQL·sql

PRAGMA user_version;
-- 0

BEGIN;
CREATE TABLE conversations (
  id INTEGER PRIMARY KEY,
  title TEXT NOT NULL,
  created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
PRAGMA user_version = 1;
COMMIT;

-- 나중에, 버전 2 가 컬럼 추가
BEGIN;
ALTER TABLE conversations ADD COLUMN brain TEXT NOT NULL DEFAULT 'claude';
PRAGMA user_version = 2;
COMMIT;

마이그레이션 runner — Python·python

import sqlite3

MIGRATIONS = [
    """CREATE TABLE conversations (
         id INTEGER PRIMARY KEY,
         title TEXT NOT NULL,
         created_at TEXT NOT NULL DEFAULT (datetime('now'))
       )""",
    "ALTER TABLE conversations ADD COLUMN brain TEXT NOT NULL DEFAULT 'claude'",
    "CREATE INDEX idx_conv_brain_created ON conversations(brain, created_at DESC)",
]

def migrate(db_path: str) -> int:
    with sqlite3.connect(db_path) as conn:
        current = conn.execute('PRAGMA user_version').fetchone()[0]
        for i, sql in enumerate(MIGRATIONS, start=1):
            if current >= i:
                continue
            conn.execute('BEGIN')
            try:
                conn.executescript(sql)
                conn.execute(f'PRAGMA user_version = {i}')
                conn.commit()
            except Exception:
                conn.rollback()
                raise
        return conn.execute('PRAGMA user_version').fetchone()[0]

print('migrated to version', migrate('myapp.db'))

Exercise

작은 마이그레이션 runner 만들기: SQL string list 받아서 순서대로 적용 + 각 후 PRAGMA user_version 증가하는 Python 함수. Fresh DB 에서 + 이미 버전 2 인 DB 에서 마이그레이션 3 개로 테스트 — runner 가 끝난 거 skip 하고 새 거만 적용해야 함. 보너스: 실패 시 롤백 transaction 으로 감싸기.

Schema 마이그레이션과 버저닝

Schema 를 코드처럼 다뤄

Code

External links

Exercise

Progress

댓글 0