BOMデータを正規化する設計｜親子関係・revision・再帰SQLの実務

{
  "bom_id": "bom-pump-unit",
  "revision": "C",
  "root_item_id": "item-pump-unit",
  "status": "released",
  "lines": [
    {
      "line_no": 10,
      "parent_item_id": "item-pump-unit",
      "child_item_id": "item-motor",
      "quantity": 1
    },
    {
      "line_no": 20,
      "parent_item_id": "item-pump-unit",
      "child_item_id": "item-bolt-m8",
      "quantity": 4
    }
  ]
}

{
  "part_number": "ASSY-001",
  "children": [
    {
      "part_number": "SUB-010",
      "quantity": 2,
      "children": []
    }
  ]
}

PRAGMA foreign_keys = ON;

CREATE TABLE items (
    item_id     TEXT PRIMARY KEY NOT NULL,
    part_number TEXT UNIQUE NOT NULL,
    name        TEXT NOT NULL CHECK (length(trim(name)) > 0),
    base_unit   TEXT NOT NULL
);

CREATE TABLE bom_revisions (
    bom_id       TEXT NOT NULL,
    revision     TEXT NOT NULL,
    root_item_id TEXT NOT NULL,
    status       TEXT NOT NULL
        CHECK (status IN ('draft', 'released', 'obsolete')),
    released_at  TEXT,
    PRIMARY KEY (bom_id, revision),
    FOREIGN KEY (root_item_id) REFERENCES items(item_id)
        ON DELETE RESTRICT
);

CREATE TABLE bom_lines (
    bom_id         TEXT NOT NULL,
    revision       TEXT NOT NULL,
    line_no        INTEGER NOT NULL CHECK (line_no > 0),
    parent_item_id TEXT NOT NULL,
    child_item_id  TEXT NOT NULL,
    quantity       INTEGER NOT NULL CHECK (quantity > 0),
    PRIMARY KEY (bom_id, revision, line_no),
    FOREIGN KEY (bom_id, revision)
        REFERENCES bom_revisions(bom_id, revision)
        ON DELETE CASCADE,
    FOREIGN KEY (parent_item_id) REFERENCES items(item_id)
        ON DELETE RESTRICT,
    FOREIGN KEY (child_item_id) REFERENCES items(item_id)
        ON DELETE RESTRICT,
    CHECK (parent_item_id <> child_item_id)
);

CREATE INDEX idx_bom_lines_parent
    ON bom_lines (bom_id, revision, parent_item_id);

released revisionを参照
  → 新しいdraft revisionを作成
  → 行を追加・変更・削除
  → 構造と循環参照を検証
  → 承認後にreleasedへ変更
  → 旧revisionは読取専用で保持

WITH RECURSIVE exploded (
    parent_item_id,
    child_item_id,
    depth,
    cumulative_quantity,
    path
) AS (
    SELECT
        parent_item_id,
        child_item_id,
        1,
        quantity,
        '/' || parent_item_id || '/' || child_item_id || '/'
    FROM bom_lines
    WHERE bom_id = :bom_id
      AND revision = :revision
      AND parent_item_id = :root_item_id

    UNION ALL

    SELECT
        bl.parent_item_id,
        bl.child_item_id,
        e.depth + 1,
        e.cumulative_quantity * bl.quantity,
        e.path || bl.child_item_id || '/'
    FROM exploded AS e
    JOIN bom_lines AS bl
      ON bl.bom_id = :bom_id
     AND bl.revision = :revision
     AND bl.parent_item_id = e.child_item_id
    WHERE instr(e.path, '/' || bl.child_item_id || '/') = 0
)
SELECT
    e.depth,
    i.part_number,
    e.cumulative_quantity
FROM exploded AS e
JOIN items AS i ON i.item_id = e.child_item_id
ORDER BY e.depth, i.part_number;

from collections import defaultdict


def validate_acyclic(lines: list[dict]) -> None:
    graph: dict[str, list[str]] = defaultdict(list)

    for line in lines:
        parent = line["parent_item_id"]
        child = line["child_item_id"]
        graph[parent].append(child)

    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(item_id: str) -> None:
        if item_id in visiting:
            raise ValueError(f"BOMに循環参照があります: {item_id}")
        if item_id in visited:
            return

        visiting.add(item_id)
        for child_id in graph[item_id]:
            visit(child_id)
        visiting.remove(item_id)
        visited.add(item_id)

    for item_id in list(graph):
        visit(item_id)

import sqlite3


def replace_draft_lines(
    conn: sqlite3.Connection,
    bom_id: str,
    revision: str,
    lines: list[dict],
) -> None:
    validate_acyclic(lines)

    with conn:
        row = conn.execute(
            """
            SELECT status
            FROM bom_revisions
            WHERE bom_id = ? AND revision = ?
            """,
            (bom_id, revision),
        ).fetchone()

        if row is None:
            raise ValueError("BOM revisionが存在しません")
        if row[0] != "draft":
            raise ValueError("released revisionは変更できません")

        conn.execute(
            "DELETE FROM bom_lines WHERE bom_id = ? AND revision = ?",
            (bom_id, revision),
        )

        conn.executemany(
            """
            INSERT INTO bom_lines (
                bom_id, revision, line_no,
                parent_item_id, child_item_id, quantity
            ) VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                (
                    bom_id,
                    revision,
                    line["line_no"],
                    line["parent_item_id"],
                    line["child_item_id"],
                    line["quantity"],
                )
                for line in lines
            ],
        )

bom_id,revision,line_no,parent_item_id,child_item_id,quantity
bom-pump-unit,C,10,item-pump-unit,item-motor,1
bom-pump-unit,C,20,item-pump-unit,item-bolt-m8,4

構文・文字コード
  → 必須fieldと型
  → IDの存在
  → revisionの更新可否
  → 数量・重複・循環
  → transactionで保存
  → 再帰queryで展開結果を確認