"""MySQL connector — the MySQL/MariaDB sibling of the Postgres connector.

Shares every bit of behaviour with ``BaseSqlConnector`` (tools, SQL safety guard,
row caps); only the PyMySQL driver seams differ. Reads run inside a session-level
READ ONLY transaction as defense in depth on top of the text guard.
"""

from __future__ import annotations

from legate.connectors.sql_base import BaseSqlConnector, _sql_config_schema


class MySQLConnector(BaseSqlConnector):
    type = "mysql"
    title = "MySQL / MariaDB"
    _default_port = 3306
    config_schema = _sql_config_schema(
        extra={"ssl": {"type": "boolean", "description": "Connect over TLS."}}
    )

    def _connect(self, config: dict, secrets: dict):  # pragma: no cover - needs a live database
        import pymysql

        kwargs: dict = {
            "host": config.get("host", ""),
            "port": int(config.get("port", self._default_port)),
            "user": config.get("user", ""),
            "password": secrets.get("password", ""),
            "database": config.get("database", ""),
        }
        if config.get("ssl"):
            kwargs["ssl"] = {}
        return pymysql.connect(**kwargs)

    def _run_query(
        self, config: dict, secrets: dict, sql: str, params: list | None, max_rows: int
    ) -> dict:  # pragma: no cover - exercised against a live database
        conn = self._connect(config, secrets)
        try:
            with conn.cursor() as cur:
                cur.execute("SET SESSION TRANSACTION READ ONLY")
                cur.execute(sql, params or None)
                columns = [d[0] for d in cur.description] if cur.description else []
                rows = [list(r) for r in cur.fetchmany(max_rows)]
            conn.rollback()
        finally:
            conn.close()
        return {"columns": columns, "rows": rows}

    def _run_exec(
        self, config: dict, secrets: dict, sql: str, params: list | None
    ) -> int:  # pragma: no cover - exercised against a live database
        conn = self._connect(config, secrets)
        try:
            with conn.cursor() as cur:
                cur.execute(sql, params or None)
                affected = cur.rowcount
            conn.commit()
        finally:
            conn.close()
        return affected
