Schema 마이그레이션과 버저닝

Schema를 코드처럼 다뤄

production SQLite schema는 시간이 지나면서 자라. 그러니 두 질문에 미리 답해둬야 해.

새로 설치한 DB를 지금 버전까지 어떻게 끌어올릴까?
이미 돌고 있는 DB를 N에서 N+1로 어떻게 안전하게 올릴까?

SQLite가 PRAGMA user_version을 줘. DB 헤더에 박혀 있는 32비트 정수인데, 이걸 마이그레이션 커서로 쓰면 돼. 피파의 backend/store/conversations.py를 비롯해 수많은 곳이 이 패턴을 써.

시작할 때 PRAGMA user_version을 읽어.
지금 버전에서 목표 버전까지, 각 마이그레이션 단계를 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 문자열 목록을 받아 순서대로 적용하고, 하나 끝날 때마다 PRAGMA user_version을 올리는 Python 함수면 돼. 마이그레이션 세 개를 준비하고 두 가지 상황에서 테스트해. 완전히 새 DB에서 한 번, 이미 버전 2까지 올라간 DB에서 한 번. runner가 이미 끝난 건 건너뛰고 새것만 적용해야 해. 보너스로, 실패하면 되돌아가도록 transaction으로 감싸봐.

Schema를 코드처럼 다뤄

Code

External links

Exercise

Progress

댓글 0