Kansho/backend/sql_compat.py

"""Translate SQLite-oriented SQL so the same stores can run against PostgreSQL.

Local Windows development and tests keep SQLite. Docker/Server set
KANSHO_DB_BACKEND=postgres. Identity, keys and journal bodies are not logged here.
"""
from __future__ import annotations

import os
import re

_INSERT_OR_IGNORE = re.compile(r"INSERT\s+OR\s+IGNORE\s+INTO", re.IGNORECASE)
_ON_CONFLICT_PAREN = re.compile(r"ON\s+CONFLICT\s*\(", re.IGNORECASE)
_PRAGMA_TABLE = re.compile(
    r"^\s*PRAGMA\s+table_info\(\s*['\"]?(\w+)['\"]?\s*\)\s*;?\s*$",
    re.IGNORECASE,
)
_SQLITE_MASTER_TABLES = re.compile(
    r"^\s*SELECT\s+name\s+FROM\s+sqlite_master\s+WHERE\s+type\s*=\s*'table'\s*;?\s*$",
    re.IGNORECASE,
)
_SQLITE_MASTER_SQL = re.compile(
    r"^\s*SELECT\s+sql\s+FROM\s+sqlite_master\s+WHERE\s+type\s*=\s*'table'\s+AND\s+name\s*=\s*\?\s*;?\s*$",
    re.IGNORECASE,
)


def use_postgres() -> bool:
    """Decide the engine at connect time, not at import time.

    Explicit KANSHO_DB_BACKEND wins. KANSHO_DB_PATH without an explicit postgres
    backend keeps tests on isolated SQLite even inside a Compose container.
    """
    raw = (os.environ.get("KANSHO_DB_BACKEND") or "").strip().lower()
    if raw in {"sqlite", "sqlite3"}:
        return False
    if raw in {"postgres", "postgresql", "pg"}:
        return True
    if (os.environ.get("KANSHO_DB_PATH") or "").strip():
        return False
    url = (os.environ.get("DATABASE_URL") or "").strip().lower()
    if url.startswith("postgres"):
        return True
    if (os.environ.get("DB_HOST") or "").strip():
        return True
    return False


def postgres_connect_kwargs() -> dict:
    url = (os.environ.get("DATABASE_URL") or "").strip()
    if url:
        return {"conninfo": url}
    host = (os.environ.get("DB_HOST") or "postgres").strip() or "postgres"
    port = int((os.environ.get("DB_PORT") or "5432").strip() or "5432")
    dbname = (os.environ.get("DB_NAME") or "kansho").strip() or "kansho"
    user = (os.environ.get("DB_USER") or "kansho").strip() or "kansho"
    password = os.environ.get("DB_PASSWORD") or ""
    return {
        "host": host,
        "port": port,
        "dbname": dbname,
        "user": user,
        "password": password,
    }


def sqlite_schema_to_postgres(sql: str) -> str:
    return sql.replace("datetime('now')", "CURRENT_TIMESTAMP::text")


def replace_placeholders(sql: str) -> str:
    """Replace SQLite `?` placeholders with psycopg `%s`, ignoring quoted text."""
    out: list[str] = []
    i = 0
    in_single = False
    in_double = False
    while i < len(sql):
        ch = sql[i]
        if ch == "'" and not in_double:
            if in_single and i + 1 < len(sql) and sql[i + 1] == "'":
                out.append("''")
                i += 2
                continue
            in_single = not in_single
            out.append(ch)
            i += 1
            continue
        if ch == '"' and not in_single:
            in_double = not in_double
            out.append(ch)
            i += 1
            continue
        if ch == "?" and not in_single and not in_double:
            out.append("%s")
            i += 1
            continue
        out.append(ch)
        i += 1
    return "".join(out)


def adapt_sql(sql: str) -> str:
    """Make a SQLite-shaped statement runnable on PostgreSQL."""
    sql = sqlite_schema_to_postgres(sql)
    sql = _ON_CONFLICT_PAREN.sub("ON CONFLICT (", sql)
    used_ignore = bool(_INSERT_OR_IGNORE.search(sql))
    sql = _INSERT_OR_IGNORE.sub("INSERT INTO", sql)
    if used_ignore and re.search(r"ON\s+CONFLICT", sql, re.IGNORECASE) is None:
        stripped = sql.rstrip()
        ended = stripped.endswith(";")
        if ended:
            stripped = stripped[:-1].rstrip()
        stripped = f"{stripped} ON CONFLICT DO NOTHING"
        sql = stripped + (";" if ended else "")
    return replace_placeholders(sql)


def rewrite_catalog_sql(sql: str) -> tuple[str, tuple | None] | None:
    """Rewrite SQLite catalog queries. Returns (sql, extra_params) or None."""
    match = _PRAGMA_TABLE.match(sql)
    if match:
        return (
            """
            SELECT column_name AS name
            FROM information_schema.columns
            WHERE table_schema = 'public' AND table_name = %s
            ORDER BY ordinal_position
            """,
            (match.group(1).lower(),),
        )
    if _SQLITE_MASTER_TABLES.match(sql):
        return (
            """
            SELECT tablename AS name
            FROM pg_catalog.pg_tables
            WHERE schemaname = 'public'
            """,
            None,
        )
    if _SQLITE_MASTER_SQL.match(sql):
        return ("SELECT NULL AS sql WHERE FALSE", None)
    return None


def split_sql(script: str) -> list[str]:
    """Split a SQL script into statements, respecting quotes."""
    statements: list[str] = []
    buf: list[str] = []
    in_single = False
    in_double = False
    i = 0
    while i < len(script):
        ch = script[i]
        if ch == "'" and not in_double:
            if in_single and i + 1 < len(script) and script[i + 1] == "'":
                buf.append("''")
                i += 2
                continue
            in_single = not in_single
            buf.append(ch)
            i += 1
            continue
        if ch == '"' and not in_single:
            in_double = not in_double
            buf.append(ch)
            i += 1
            continue
        if ch == ";" and not in_single and not in_double:
            stmt = "".join(buf).strip()
            if stmt:
                statements.append(stmt)
            buf = []
            i += 1
            continue
        buf.append(ch)
        i += 1
    tail = "".join(buf).strip()
    if tail:
        statements.append(tail)
    return statements