스키마 진화와 이전 버전 호환성

스키마는 시간이 지나면서 왜 변경되는가

스키마 버전 관리하기

from pydantic import BaseModel
from typing import Literal

class InvoiceV1(BaseModel):
    schema_version: Literal['1.0'] = '1.0'
    vendor: str
    total_amount: float

class InvoiceV2(BaseModel):
    schema_version: Literal['2.0'] = '2.0'
    vendor: str
    vendor_tax_id: str | None = None  # new field
    total_amount: float
    currency: str = 'USD'  # new field with default

추가 변경과 호환성을 깨는 변경

# Safe: additive change - add optional field
class ProductV2(BaseModel):
    name: str
    price: float
    sku: str | None = None   # NEW optional field - backward safe
    category: str = 'general'  # NEW with default - backward safe

# Risky: breaking change - rename or retype
# class ProductV2(BaseModel):
#     product_name: str  # RENAMED from name - breaks consumers
#     price_cents: int   # RETYPED from float - breaks data

데이터베이스에 스키마 버전 저장하기

-- PostgreSQL table design
CREATE TABLE extractions (
    doc_id TEXT PRIMARY KEY,
    schema_version TEXT NOT NULL,
    extracted_data JSONB NOT NULL,
    extracted_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_schema_version ON extractions(schema_version);

-- Query old records needing migration
SELECT doc_id, extracted_data
FROM extractions
WHERE schema_version = '1.0'
LIMIT 1000;

마이그레이션 스크립트 작성하기

import asyncpg
import json

async def migrate_v1_to_v2(pool, batch_size=100):
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            'SELECT doc_id, extracted_data FROM extractions WHERE schema_version=$1 LIMIT $2',
            '1.0', batch_size
        )
        for row in rows:
            old = row['extracted_data']
            new_data = {
                'schema_version': '2.0',
                'vendor': old['vendor'],
                'vendor_tax_id': None,  # unknown for old records
                'total_amount': old['total_amount'],
                'currency': 'USD'  # assume USD for old records
            }
            await conn.execute(
                'UPDATE extractions SET extracted_data=$1, schema_version=$2 WHERE doc_id=$3',
                json.dumps(new_data), '2.0', row['doc_id']
            )

전환 중 병렬 유효성 검사

async def parallel_validate(text: str) -> dict:
    v1_result, v2_result = await asyncio.gather(
        extract_with_schema(text, InvoiceV1),
        extract_with_schema(text, InvoiceV2)
    )
    discrepancy = (
        v1_result.vendor != v2_result.vendor or
        abs(v1_result.total_amount - v2_result.total_amount) > 0.01
    )
    if discrepancy:
        log_discrepancy(text, v1_result, v2_result)
    return {'v1': v1_result, 'v2': v2_result, 'discrepancy': discrepancy}

스키마 출시를 위한 기능 플래그

import os

def get_active_schema():
    version = os.environ.get('EXTRACTION_SCHEMA_VERSION', '1.0')
    schemas = {
        '1.0': InvoiceV1,
        '2.0': InvoiceV2,
    }
    return schemas.get(version, InvoiceV1)

async def extract_document(text: str):
    SchemaClass = get_active_schema()
    return await extract_with_schema(text, SchemaClass)

유니온 형식으로 소비자 호환성 유지하기

from pydantic import BaseModel
from typing import Union, Annotated
from typing import Literal

def parse_extraction(raw: dict) -> Union[InvoiceV1, InvoiceV2]:
    version = raw.get('schema_version', '1.0')
    if version == '1.0':
        return InvoiceV1(**raw)
    elif version == '2.0':
        return InvoiceV2(**raw)
    else:
        raise ValueError(f'Unknown schema version: {version}')

배포 전에 스키마 변경 테스트하기

def eval_schema_on_test_set(test_cases: list, SchemaClass) -> dict:
    field_f1 = {}
    for case in test_cases:
        result = extract_with_schema(case['text'], SchemaClass)
        for field in case['expected']:
            expected = case['expected'][field]
            actual = getattr(result, field, None)
            # Update precision/recall counters
            update_metrics(field_f1, field, expected, actual)
    return {k: compute_f1(v) for k, v in field_f1.items()}

스키마 지원 중단 처리하기

DEPRECATED_VERSIONS = {'1.0'}
SUNSET_DATE = '2026-09-01'

def warn_if_deprecated(version: str):
    if version in DEPRECATED_VERSIONS:
        import warnings
        warnings.warn(
            f'Schema version {version} is deprecated. '
            f'It will be removed after {SUNSET_DATE}. '
            'Migrate consumers to version 2.0.',
            DeprecationWarning,
            stacklevel=2
        )

변경 기록과 소통

# CHANGELOG.md entry format:
# ## Schema v2.0 (2026-07-01)
# ### Changes
# - ADDED: vendor_tax_id (Optional[str]) - VAT/EIN extracted from header
# - ADDED: currency (str, default='USD') - detected from symbol/code
# ### Migration
# Run: python scripts/migrate_v1_to_v2.py --batch-size=500
# ### Consumers
# - billing-service: update parse_extraction() to handle v2
# - audit-service: query now supports currency filter

빠른 확인

레슨 요약