Forge-Engine/game_engine/flow/add_flow_node.py

from typing import Any

import db

from .constants import FLOW_NODE_FIELDS
from .ensure_flow_schema import ensure_flow_schema


def add_flow_node(slug: str, screen_id: int, node_type: str, x: float = 40, y: float = 40, **fields: Any) -> int:
    ensure_flow_schema(slug)
    cols: list[str] = ["screen_id", "node_type", "x", "y"]
    vals: list[Any] = [screen_id, node_type, x, y]
    for k, v in fields.items():
        if k in FLOW_NODE_FIELDS and v not in (None, ""):
            cols.append(k)
            vals.append(v)
    placeholders = ", ".join("?" for _ in cols)
    conn = db.connect(slug)
    conn.execute(f"INSERT INTO _flow_nodes ({', '.join(cols)}) VALUES ({placeholders})", vals)  # nosec B608  # noqa: S608 - cols = liste fixe + cles filtrees contre FLOW_NODE_FIELDS (whitelist), jamais une cle arbitraire ; valeurs parametrees (?)
    node_id = int(conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"])
    conn.commit()
    conn.close()
    return node_id