SQLite의 NULL과 빈 문자열을 CSV로 내보낼 때 구분하기

#!/usr/bin/env python3
"""Five-row SQLite NULL/TEXT CSV exercise. Python standard library only.

This is a teaching example for a closed, synthetic database, not a backup
tool for a live database. Read README.md before using the commands.
"""
import argparse
import csv
import hashlib
import os
from pathlib import Path
import sqlite3
import sys

HEADER = ['id', 'value', 'null_flag']
DEMO_ROWS = [
    ('N01', None),
    ('N02', ''),
    ('N03', 'NULL'),
    ('N04', '0'),
    ('N05', 'text'),
]
ROW_COUNT = 5
MAX_FIELD_CHARS = 131_072  # Same bound for export and strict CSV import.
DEMO_SUMMARY = {'rows': 5, 'null': 1, 'empty_text': 1, 'nonempty_text': 3}
SCHEMA = 'CREATE TABLE null_demo (id TEXT PRIMARY KEY NOT NULL, value)'


class ExerciseError(ValueError):
    """Input does not satisfy this small exercise's contract."""


def _source_path(path):
    path = Path(path).resolve()
    if not path.is_file():
        raise ExerciseError(f'source file does not exist or is not a file: {path}')
    return path


def _new_output(path):
    path = Path(path).absolute()
    # lexists also rejects a dangling symlink, which must not be replaced.
    if os.path.lexists(path):
        raise ExerciseError(f'output already exists; choose a new path: {path}')
    return path


def _different_paths(first, second):
    first, second = Path(first), Path(second)
    same = first.resolve() == second.resolve()
    if first.exists() and second.exists():
        same = same or os.path.samefile(first, second)
    if same:
        raise ExerciseError('source and destination must be different paths')


def _sha256(path):
    digest = hashlib.sha256()
    with Path(path).open('rb') as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b''):
            digest.update(chunk)
    return digest.hexdigest()


def _snapshot(path):
    """Read the five rows through mode=ro and check NULL/TEXT storage types."""
    path = _source_path(path)
    before = _sha256(path)
    connection = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True)
    try:
        records = connection.execute(
            'SELECT id, value, typeof(value), typeof(id) '
            'FROM null_demo ORDER BY id'
        ).fetchall()
    finally:
        connection.close()
    if _sha256(path) != before:
        raise ExerciseError('database SHA-256 changed during read; stop the exercise')
    if len(records) != ROW_COUNT:
        raise ExerciseError(f'exactly {ROW_COUNT} rows required; got {len(records)}')
    seen = set()
    rows = []
    for identifier, value, value_type, id_type in records:
        if id_type != 'text' or not isinstance(identifier, str) or not identifier:
            raise ExerciseError('id must be a nonempty TEXT value')
        if identifier in seen:
            raise ExerciseError(f'duplicate id: {identifier!r}')
        seen.add(identifier)
        if value_type not in ('null', 'text'):
            raise ExerciseError(
                f'{identifier}: unsupported SQLite type {value_type}; '
                'INTEGER, REAL and BLOB are outside this NULL/TEXT exercise'
            )
        if len(identifier) > MAX_FIELD_CHARS or (
            value_type == 'text' and len(value) > MAX_FIELD_CHARS
        ):
            raise ExerciseError(f'{identifier[:40]}: CSV fields may contain at most {MAX_FIELD_CHARS} characters')
        rows.append((identifier, value, value_type))
    return rows, before


def _write_new_database(target, rows):
    """Exclusive new file; explicit transaction covers schema and all inserts."""
    target = _new_output(target)
    try:
        descriptor = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    except FileExistsError as error:
        raise ExerciseError(f'output already exists: {target}') from error
    os.close(descriptor)
    try:
        # mode=rw requires the file reserved above; it does not create another file.
        connection = sqlite3.connect(target.resolve().as_uri() + '?mode=rw', uri=True)
        try:
            connection.execute('BEGIN')
            connection.execute(SCHEMA)
            for identifier, value in rows:
                # Binding None creates SQL NULL. Binding str keeps even '0' TEXT
                # because value has no declared type / no numeric affinity.
                connection.execute(
                    'INSERT INTO null_demo (id, value) VALUES (?, ?)',
                    (identifier, value),
                )
            connection.commit()
        except BaseException:
            connection.rollback()
            raise
        finally:
            connection.close()
    except BaseException:
        # Only the new file reserved by this invocation is removed.
        # The exercise assumes nobody concurrently changes these paths.
        target.unlink(missing_ok=True)
        raise


def create_demo(source):
    """Create a NEW synthetic source database; never overwrite one."""
    _write_new_database(source, DEMO_ROWS)


def export_csv(source, target):
    """Export NULL as empty+1, TEXT as its exact value+0; source is read-only."""
    _different_paths(source, target)
    target = _new_output(target)
    rows, source_sha = _snapshot(source)
    created = False
    try:
        with target.open('x', encoding='utf-8', errors='strict', newline='') as handle:
            created = True
            writer = csv.writer(handle, lineterminator='\n')
            writer.writerow(HEADER)
            for identifier, value, value_type in rows:
                writer.writerow([
                    identifier,
                    '' if value_type == 'null' else value,
                    '1' if value_type == 'null' else '0',
                ])
        if _sha256(_source_path(source)) != source_sha:
            raise ExerciseError('source SHA-256 changed during export; stop the exercise')
    except BaseException:
        if created:
            target.unlink(missing_ok=True)
        raise
    return source_sha


def _read_csv(source):
    rows = []
    seen = set()
    previous_limit = csv.field_size_limit(MAX_FIELD_CHARS)
    try:
        with source.open('r', encoding='utf-8', errors='strict', newline='') as handle:
            reader = csv.reader(handle, strict=True)
            if next(reader, None) != HEADER:
                raise ExerciseError('CSV header must be exactly id,value,null_flag in that order')
            for record in reader:
                if len(record) != len(HEADER):
                    raise ExerciseError(f'CSV record ending at line {reader.line_num}: exactly 3 fields required')
                identifier, value, flag = record
                if not identifier or identifier in seen:
                    raise ExerciseError(f'CSV id must be nonempty and unique: {identifier!r}')
                seen.add(identifier)
                if flag not in ('0', '1'):
                    raise ExerciseError(f'{identifier}: null_flag must be exactly 0 or 1')
                if flag == '1' and value != '':
                    raise ExerciseError(f'{identifier}: null_flag=1 requires an empty value field')
                rows.append((identifier, None if flag == '1' else value))
    except (UnicodeError, csv.Error) as error:
        raise ExerciseError(f'invalid UTF-8 or malformed CSV: {error}') from error
    finally:
        csv.field_size_limit(previous_limit)
    if len(rows) != ROW_COUNT:
        raise ExerciseError(f'exactly {ROW_COUNT} CSV data rows required; got {len(rows)}')
    return rows


def restore_csv(source, target):
    """Validate all CSV first, then restore into a NEW database transaction."""
    _different_paths(source, target)
    source = _source_path(source)
    target = _new_output(target)
    rows = _read_csv(source)
    _write_new_database(target, rows)


def verify(source, restored):
    """Compare id -> (value, typeof(value)), then enforce the demo's counts."""
    _different_paths(source, restored)
    source_rows, _ = _snapshot(source)
    restored_rows, _ = _snapshot(restored)
    expected = {identifier: (value, kind) for identifier, value, kind in source_rows}
    actual = {identifier: (value, kind) for identifier, value, kind in restored_rows}
    if expected.keys() != actual.keys():
        raise ExerciseError('source and restored id sets differ')
    for identifier in sorted(expected):
        if expected[identifier] != actual[identifier]:
            raise ExerciseError(
                f'{identifier}: value/type mismatch: '
                f'source={expected[identifier]!r}; restored={actual[identifier]!r}'
            )
    summary = {
        'rows': len(restored_rows),
        'null': sum(kind == 'null' for _, _, kind in restored_rows),
        'empty_text': sum(kind == 'text' and value == '' for _, value, kind in restored_rows),
        'nonempty_text': sum(kind == 'text' and value != '' for _, value, kind in restored_rows),
    }
    if summary != DEMO_SUMMARY:
        raise ExerciseError(f'exercise counts differ: expected {DEMO_SUMMARY}; got {summary}')
    return summary


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest='command', required=True)
    create = commands.add_parser('create-demo', help='create a new five-row source database')
    create.add_argument('source', type=Path)
    for name, help_text in (
        ('export', 'read-only source database to a new UTF-8 CSV'),
        ('restore', 'validated CSV to a new database'),
        ('verify', 'read-only comparison of two databases'),
    ):
        command = commands.add_parser(name, help=help_text)
        command.add_argument('source', type=Path)
        command.add_argument('target', type=Path)
    args = parser.parse_args(argv)
    try:
        if args.command == 'create-demo':
            create_demo(args.source)
            print('OK create-demo: 5 rows in new source database')
        elif args.command == 'export':
            source_sha = export_csv(args.source, args.target)
            print(f'OK export: 5 rows; source_sha256={source_sha}')
        elif args.command == 'restore':
            restore_csv(args.source, args.target)
            print('OK restore: 5 rows committed in one transaction to new database')
        else:
            verify(args.source, args.target)
            print('OK verify: 5 rows = 1 NULL + 1 empty TEXT + 3 nonempty TEXT; values/types match by id')
    except (OSError, ValueError, sqlite3.Error, csv.Error) as error:
        print(f'ERROR: {error}', file=sys.stderr)
        return 2
    return 0


if __name__ == '__main__':
    raise SystemExit(main())