JSON to SQL INSERT — Python source

"""json-to-sql — Pure JSON → SQL INSERT converter.
Language: Python (3.10+).

CosmoDev polyglot showcase port of the `json-to-sql` tool, ported from
src/lib/jsonToSql.ts. Display source — part of CosmoDev's polyglot tool
pages (https://dev.cosmolabs.org).

The converter never raises on user input: malformed JSON is reported via the
returned Result's ``error`` field, matching the TypeScript contract. Python
dicts preserve insertion order, so first-seen column ordering matches the
reference implementation exactly.
"""

from __future__ import annotations

import json
import math
import re
from typing import Any, Literal, TypedDict

Dialect = Literal["standard", "mysql", "postgres"]


class Options(TypedDict, total=False):
    """INSERT builder configuration. ``table`` is required at call sites."""
    table: str
    dialect: Dialect
    quoteIdentifiers: bool


class Result(TypedDict):
    """Conversion outcome. ``error`` is None when ``ok`` is True."""
    ok: bool
    sql: str
    rows: int
    error: str | None


def escape_sql_string(s: str, dialect: Dialect = "standard") -> str:
    """Escape a string for a single-quoted SQL literal.

    Single quotes are always doubled; MySQL mode additionally escapes the
    backslash and the control bytes that terminate a MySQL statement.
    """
    out = s.replace("'", "''")
    if dialect == "mysql":
        # Backslash first, so the backslashes added below aren't re-doubled
        # (matches the canonical TypeScript replacement order).
        out = out.replace("\\", "\\\\")   # \   -> \\
        out = out.replace("\0", "\\0")    # NUL -> \0
        out = out.replace("\n", "\\n")    # LF  -> \n
        out = out.replace("\r", "\\r")    # CR  -> \r
        out = out.replace("\x1a", "\\Z")  # SUB -> \Z (ends MySQL input)
    return out


def _format_number(n: int | float) -> str:
    """Mimic JavaScript's ``String(number)``: integer-valued floats drop the
    trailing ``.0`` (``1.0`` -> ``"1"``) so literals match the TS port."""
    if isinstance(n, float) and n.is_integer() and abs(n) < 1e16:
        return str(int(n))
    return str(n)


def _normalize_for_json(value: Any) -> Any:
    """Recursively coerce integer-valued floats to ints before json.dumps so
    nested JSON text literals match JavaScript's ``JSON.stringify`` number
    rendering (which has no ``1.0`` form). Booleans pass through unchanged."""
    if isinstance(value, bool):
        return value
    if isinstance(value, float):
        if value.is_integer() and abs(value) < 1e16:
            return int(value)
        return value
    if isinstance(value, dict):
        return {k: _normalize_for_json(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_normalize_for_json(v) for v in value]
    return value


def _stringify(value: Any) -> str:
    """Serialize a nested value to compact JSON, matching JavaScript's
    ``JSON.stringify``: no whitespace, ``/`` and non-ASCII left unescaped."""
    return json.dumps(
        _normalize_for_json(value),
        separators=(",", ":"),
        ensure_ascii=False,
    )


def sql_literal(value: Any, dialect: Dialect) -> str:
    """Render a decoded JSON value as a SQL literal."""
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        # bool MUST be checked before int — in Python bool subclasses int.
        # MySQL has no BOOL literal (TINYINT(1)); standard/Postgres use keywords.
        if dialect == "mysql":
            return "1" if value else "0"
        return "TRUE" if value else "FALSE"
    if isinstance(value, (int, float)):
        # JSON never parses to NaN/Infinity, but guard for direct callers.
        if isinstance(value, float) and not math.isfinite(value):
            return "NULL"
        return _format_number(value)
    if isinstance(value, str):
        return f"'{escape_sql_string(value, dialect)}'"
    # Objects (dict) / arrays (list) → compact JSON text literal.
    return f"'{escape_sql_string(_stringify(value), dialect)}'"


def _quote_ident(name: str, dialect: Dialect, quote_identifiers: bool) -> str:
    """Wrap an identifier in dialect-appropriate quotes, or leave it bare
    when identifier quoting is disabled."""
    if not quote_identifiers:
        return name
    if dialect == "mysql":
        return f"`{name}`"
    return f'"{name}"'


def _sanitize_ident(name: str) -> str:
    """Reduce an identifier to [A-Za-z0-9_], defaulting to ``tbl`` when
    nothing usable remains. Runs before quoting, even when quoting is off."""
    cleaned = re.sub(r"[^A-Za-z0-9_]", "_", name)
    return cleaned or "tbl"


def json_to_insert(json_string: str, opts: Options) -> Result:
    """Convert a JSON document into a single multi-row INSERT statement.

    A bare JSON object is treated as one row; a JSON array as many. Rows may
    have differing shapes — the column list is the union of every key in
    first-seen order, and missing values are emitted as NULL.
    """
    try:
        data = json.loads(json_string)
    except json.JSONDecodeError as e:
        return {"ok": False, "sql": "", "rows": 0, "error": e.msg}

    # A bare value (object/scalar/null) wraps as a single row.
    rows = data if isinstance(data, list) else [data]
    if len(rows) == 0:
        return {"ok": False, "sql": "", "rows": 0, "error": "No rows to insert."}
    if not all(isinstance(r, dict) for r in rows):
        return {"ok": False, "sql": "", "rows": 0, "error": "Rows must be objects."}

    dialect: Dialect = opts.get("dialect", "standard")
    quote_identifiers: bool = opts.get("quoteIdentifiers", True)
    table = _quote_ident(_sanitize_ident(opts.get("table", "")), dialect, quote_identifiers)

    # Union of every row's keys, first-seen order. dict preserves insertion
    # order, so this mirrors Object.keys() across heterogeneous rows.
    cols: list[str] = []
    for row in rows:
        for k in row.keys():
            if k not in cols:
                cols.append(k)
    col_list = ", ".join(_quote_ident(c, dialect, quote_identifiers) for c in cols)

    value_lines = []
    for row in rows:
        vals = ", ".join(
            sql_literal(row[c] if c in row else None, dialect) for c in cols
        )
        value_lines.append(f"  ({vals})")

    sql = f"INSERT INTO {table} ({col_list}) VALUES\n" + ",\n".join(value_lines) + ";"
    return {"ok": True, "sql": sql, "rows": len(rows), "error": None}