CSV一括取込を壊さないstaging table設計｜検証・エラー隔離・UPSERT

CSV file
  → raw staging
  → 型変換・検証
  → valid / invalidに分類
  → transactionで本番へ反映
  → 取込結果を記録

source_record_id,part_number,name,mass_kg,active
00125,P-001,Bracket,0.50,true
00126,P-002,Shaft,abc,false

{
  "import_id": "imp-20261004-001",
  "row_number": 3,
  "raw": {
    "source_record_id": "00126",
    "part_number": "P-002",
    "name": "Shaft",
    "mass_kg": "abc",
    "active": "false"
  },
  "validation_status": "invalid",
  "errors": [
    {
      "field": "mass_kg",
      "code": "invalid_decimal"
    }
  ]
}

PRAGMA foreign_keys = ON;

CREATE TABLE import_runs (
    import_id       TEXT PRIMARY KEY NOT NULL,
    source_system   TEXT NOT NULL,
    source_filename TEXT NOT NULL,
    file_sha256     TEXT NOT NULL,
    schema_version  INTEGER NOT NULL CHECK (schema_version > 0),
    status          TEXT NOT NULL CHECK (
        status IN ('received', 'validating', 'ready', 'applied', 'rejected')
    ),
    received_at     TEXT NOT NULL,
    applied_at      TEXT,
    UNIQUE (source_system, file_sha256, schema_version)
);

CREATE TABLE part_staging (
    import_id            TEXT NOT NULL,
    row_number           INTEGER NOT NULL CHECK (row_number >= 2),
    raw_source_record_id TEXT,
    raw_part_number      TEXT,
    raw_name             TEXT,
    raw_mass_kg          TEXT,
    raw_active           TEXT,
    normalized_part_no   TEXT,
    normalized_name      TEXT,
    mass_grams           INTEGER,
    active_value         INTEGER,
    validation_status    TEXT NOT NULL CHECK (
        validation_status IN ('pending', 'valid', 'invalid')
    ),
    error_codes_json     TEXT,
    PRIMARY KEY (import_id, row_number),
    FOREIGN KEY (import_id) REFERENCES import_runs(import_id)
        ON DELETE CASCADE
);

CREATE TABLE parts (
    source_system   TEXT NOT NULL,
    source_record_id TEXT NOT NULL,
    part_number     TEXT NOT NULL UNIQUE,
    name            TEXT NOT NULL CHECK (length(trim(name)) > 0),
    mass_grams      INTEGER NOT NULL CHECK (mass_grams >= 0),
    active          INTEGER NOT NULL CHECK (active IN (0, 1)),
    updated_at      TEXT NOT NULL,
    PRIMARY KEY (source_system, source_record_id)
) STRICT;

import csv
from decimal import Decimal, InvalidOperation
from pathlib import Path


EXPECTED_FIELDS = [
    "source_record_id",
    "part_number",
    "name",
    "mass_kg",
    "active",
]


def validate_row(row: dict[str, str | None]) -> tuple[dict, list[str]]:
    errors: list[str] = []

    source_record_id = (row.get("source_record_id") or "").strip()
    part_number = (row.get("part_number") or "").strip()
    name = (row.get("name") or "").strip()

    if not source_record_id:
        errors.append("source_record_id.required")
    if not part_number:
        errors.append("part_number.required")
    if not name:
        errors.append("name.required")

    mass_grams = None
    try:
        mass_kg = Decimal(row.get("mass_kg") or "")
        if not mass_kg.is_finite() or mass_kg < 0:
            raise ValueError
        grams = mass_kg * Decimal("1000")
        if grams != grams.to_integral_value():
            errors.append("mass_kg.too_precise")
        else:
            mass_grams = int(grams)
    except (InvalidOperation, ValueError):
        errors.append("mass_kg.invalid_decimal")

    active_text = (row.get("active") or "").strip().lower()
    active_value = {"true": 1, "false": 0}.get(active_text)
    if active_value is None:
        errors.append("active.invalid_boolean")

    normalized = {
        "source_record_id": source_record_id,
        "part_number": part_number,
        "name": name,
        "mass_grams": mass_grams,
        "active": active_value,
    }
    return normalized, errors


def read_csv_rows(path: Path):
    with path.open("r", encoding="utf-8", newline="") as file:
        reader = csv.DictReader(file)
        if reader.fieldnames != EXPECTED_FIELDS:
            raise ValueError("CSV headerが契約と一致しません")

        for row in reader:
            normalized, errors = validate_row(row)
            yield reader.line_num, row, normalized, errors

INSERT INTO parts (
    source_system,
    source_record_id,
    part_number,
    name,
    mass_grams,
    active,
    updated_at
)
SELECT
    r.source_system,
    s.raw_source_record_id,
    s.normalized_part_no,
    s.normalized_name,
    s.mass_grams,
    s.active_value,
    CURRENT_TIMESTAMP
FROM part_staging AS s
JOIN import_runs AS r ON r.import_id = s.import_id
WHERE s.import_id = :import_id
  AND s.validation_status = 'valid'
ON CONFLICT (source_system, source_record_id)
DO UPDATE SET
    part_number = excluded.part_number,
    name = excluded.name,
    mass_grams = excluded.mass_grams,
    active = excluded.active,
    updated_at = excluded.updated_at;

import sqlite3


def apply_import(conn: sqlite3.Connection, import_id: str) -> None:
    invalid_count = conn.execute(
        """
        SELECT count(*)
        FROM part_staging
        WHERE import_id = ? AND validation_status <> 'valid'
        """,
        (import_id,),
    ).fetchone()[0]

    if invalid_count:
        raise ValueError(f"不正行が{invalid_count}件あります")

    with conn:
        conn.execute(UPSERT_SQL, {"import_id": import_id})
        conn.execute(
            """
            UPDATE import_runs
            SET status = 'applied',
                applied_at = CURRENT_TIMESTAMP
            WHERE import_id = ? AND status = 'ready'
            """,
            (import_id,),
        )

{
  "import_id": "imp-20261004-001",
  "status": "rejected",
  "total_rows": 2,
  "valid_rows": 1,
  "invalid_rows": 1,
  "errors": [
    {
      "row_number": 3,
      "field": "mass_kg",
      "code": "invalid_decimal"
    }
  ]
}