Harden dialect: narrow exception handling, URL query params, explicit AUTOCOMMIT rejection

ISSUE-16 — do_ping now catches only connection-level exceptions
(OperationalError, InterfaceError, DqliteConnectionError, OSError,
TimeoutError). Unexpected errors (bugs, MemoryError, etc.) propagate
so the pool doesn't silently discard a healthy connection as "dead".

ISSUE-18 — is_disconnect now dispatches on exception type first
(DqliteConnectionError, ConnectionError/BrokenPipeError/OSError,
leader-change codes 10250/10506), falling back to the substring list.
The hand-maintained list still exists as a last resort for error
kinds that aren't yet modelled as specific exception types.

ISSUE-19 — create_connect_args parses URL query params with a
strict allowlist. ?timeout=5.0 plumbs through to dqlitedbapi.connect;
unknown keys raise ArgumentError so typos surface instead of being
silently dropped.

ISSUE-21 — set_isolation_level now raises ArgumentError for
AUTOCOMMIT with an explanation. Other unsupported levels still emit
a warning (future-proof).

ISSUE-23 — supported_isolation_levels = ("SERIALIZABLE",) declared
explicitly. Applications can introspect via engine.dialect.

ISSUE-24 — dqlitedbapi.exceptions / dqliteclient.exceptions imports
moved to module top; no circular import was actually blocking them.

Existing tests that asserted the old blanket-Exception behavior
(test_ping_returns_false_on_error, test_ping_closes_cursor_even_on_error,
test_returns_fallback_on_error) were rewritten to reflect the new
narrow-exception contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -1,12 +1,16 @@
 """Base dqlite dialect for SQLAlchemy."""
 
 import datetime
+import warnings
 from typing import Any
 
+import dqliteclient.exceptions as _client_exc
+import dqlitedbapi.exceptions as _dbapi_exc
 from sqlalchemy import types as sqltypes
 from sqlalchemy.dialects.sqlite.base import SQLiteDialect
 from sqlalchemy.engine import URL
 from sqlalchemy.engine.interfaces import DBAPIConnection, IsolationLevel
+from sqlalchemy.exc import ArgumentError
 
 
 class _DqliteDateTime(sqltypes.DateTime):
@@ -57,6 +61,11 @@ class DqliteDialect(SQLiteDialect):
     # Enable SQLAlchemy statement caching
     supports_statement_cache = True
 
+    # dqlite runs every statement through Raft consensus; there is no
+    # exposed way to weaken isolation. Declaring this explicitly lets
+    # applications introspect via ``engine.dialect.supported_isolation_levels``.
+    supported_isolation_levels: tuple[str, ...] = ("SERIALIZABLE",)
+
     # Override the SQLite dialect's string-based DATE/DATETIME processors:
     # dqlitedbapi returns datetime objects (PEP 249), not ISO strings.
     colspecs = {
@@ -71,21 +80,42 @@ def import_dbapi(cls) -> Any:
 
         return dqlitedbapi
 
+    # Whitelist of URL query parameters we forward to the DBAPI connect
+    # call. Unknown keys raise ``ArgumentError`` so typos surface.
+    _URL_QUERY_ALLOWED: dict[str, type] = {"timeout": float}
+
     def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
         """Create connection arguments from URL.
 
-        URL format: dqlite://host:port/database
+        URL format: ``dqlite://host:port/database?timeout=...``
+
+        Known query parameters are typed and passed through to the
+        DBAPI. Unknown ones raise :class:`ArgumentError` so typos like
+        ``?timeoutt=5`` don't silently disappear.
         """
         host = url.host or "localhost"
         port = url.port or 9001
         database = url.database or "default"
 
         address = f"{host}:{port}"
-
-        return [], {
-            "address": address,
-            "database": database,
-        }
+        kwargs: dict[str, Any] = {"address": address, "database": database}
+
+        query = dict(url.query) if url.query else {}
+        for key, raw in query.items():
+            if key not in self._URL_QUERY_ALLOWED:
+                raise ArgumentError(
+                    f"Unknown dqlite URL query parameter {key!r}. "
+                    f"Allowed: {sorted(self._URL_QUERY_ALLOWED)}"
+                )
+            converter = self._URL_QUERY_ALLOWED[key]
+            try:
+                kwargs[key] = converter(raw)
+            except (TypeError, ValueError) as e:
+                raise ArgumentError(
+                    f"Cannot convert URL query {key}={raw!r} to {converter.__name__}: {e}"
+                ) from e
+
+        return [], kwargs
 
     def get_isolation_level(self, dbapi_connection: DBAPIConnection) -> IsolationLevel:
         """Return the isolation level.
@@ -98,17 +128,25 @@ def get_isolation_level(self, dbapi_connection: DBAPIConnection) -> IsolationLev
     def set_isolation_level(self, dbapi_connection: DBAPIConnection, level: str | None) -> None:
         """Set isolation level.
 
-        dqlite only supports SERIALIZABLE isolation. A warning is emitted
-        if a different level is requested.
+        dqlite only supports SERIALIZABLE. ``AUTOCOMMIT`` is explicitly
+        rejected because silently dropping the request would cause users
+        to lose transactionality without knowing it. Other unsupported
+        levels emit a warning (future-proof for isolation levels dqlite
+        may grow to support).
         """
-        if level is not None and level != "SERIALIZABLE":
-            import warnings
-
-            warnings.warn(
-                f"dqlite only supports SERIALIZABLE isolation. "
-                f"Requested level {level!r} is ignored.",
-                stacklevel=2,
+        if level is None or level == "SERIALIZABLE":
+            return
+        if level == "AUTOCOMMIT":
+            raise ArgumentError(
+                "dqlite does not support AUTOCOMMIT; every statement goes through "
+                "Raft consensus and there is no per-statement autocommit mode. "
+                "Use explicit commit() / rollback() on the connection."
             )
+        warnings.warn(
+            f"dqlite only supports SERIALIZABLE isolation. "
+            f"Requested level {level!r} is ignored.",
+            stacklevel=2,
+        )
 
     # do_rollback / do_commit are intentionally left inherited from the
     # parent dialect. The "cannot commit/rollback — no transaction is
@@ -124,31 +162,67 @@ def set_isolation_level(self, dbapi_connection: DBAPIConnection, level: str | No
         "Not connected",
     )
 
+    # Server-side SQLite error codes that mean the connection is useless
+    # even if the TCP socket is still alive — kept in sync with the
+    # client's _LEADER_ERROR_CODES.
+    #   SQLITE_IOERR_NOT_LEADER       = SQLITE_IOERR | (40 << 8) = 10250
+    #   SQLITE_IOERR_LEADERSHIP_LOST  = SQLITE_IOERR | (41 << 8) = 10506
+    _LEADER_CHANGE_CODES: frozenset[int] = frozenset({10250, 10506})
+
     def is_disconnect(self, e: Any, connection: Any, cursor: Any) -> bool:
         """Detect whether an exception indicates a broken connection.
 
-        dqlite is a network database, so we must detect TCP-level and
-        leader-change errors that the inherited pysqlite patterns miss.
+        Prefer exception-type dispatch over message matching; the C
+        server's error wording is not a contract. Type-based checks
+        cover TCP resets, DNS failures, and partial-read timeouts that
+        the hand-maintained substring list misses.
         """
-        import dqlitedbapi.exceptions
-
-        if isinstance(e, dqlitedbapi.exceptions.OperationalError):
+        # Explicit connection-level error types from the client layer.
+        if isinstance(e, _client_exc.DqliteConnectionError):
+            return True
+        # Underlying OS-level transport failures (socket RST, broken pipe,
+        # DNS, connect refused) surface as these stdlib types.
+        if isinstance(e, (ConnectionError, BrokenPipeError, TimeoutError, OSError)):
+            return True
+        # Leader-change error codes signal that the connection is useless
+        # even though it's TCP-alive.
+        for err in (_dbapi_exc.OperationalError, _client_exc.OperationalError):
+            if isinstance(e, err) and getattr(e, "code", None) in self._LEADER_CHANGE_CODES:
+                return True
+        # Legacy substring fallback — kept so we still catch anything
+        # that wasn't modelled as a specific exception type yet.
+        if isinstance(e, _dbapi_exc.OperationalError):
             msg = str(e)
             for pattern in self._dqlite_disconnect_messages:
                 if pattern in msg:
                     return True
         return super().is_disconnect(e, connection, cursor)
 
     def do_ping(self, dbapi_connection: Any) -> bool:
-        """Check if the connection is still alive."""
+        """Check if the connection is still alive.
+
+        Only connection-level exceptions are interpreted as "dead"; any
+        other exception propagates so the caller can see real bugs
+        instead of having them silently rewritten as "please reconnect."
+        """
         cursor = dbapi_connection.cursor()
         try:
-            cursor.execute("SELECT 1")
-            return True
-        except Exception:
-            return False
+            try:
+                cursor.execute("SELECT 1")
+                return True
+            except (
+                _dbapi_exc.OperationalError,
+                _dbapi_exc.InterfaceError,
+                _client_exc.DqliteConnectionError,
+                OSError,
+                TimeoutError,
+            ):
+                return False
         finally:
-            cursor.close()
+            try:
+                cursor.close()
+            except Exception:
+                pass  # cursor close shouldn't crash the ping result
 
     def _get_server_version_info(self, connection: Any) -> tuple[int, ...]:
         """Return the server version as a tuple.
@@ -160,7 +234,10 @@ def _get_server_version_info(self, connection: Any) -> tuple[int, ...]:
             version_str = result.scalar()
             if version_str:
                 return tuple(int(x) for x in version_str.split("."))
-        except Exception:
+        except (
+            _dbapi_exc.OperationalError,
+            _client_exc.DqliteConnectionError,
+        ):
             pass
         return (3, 0, 0)
 

tests/test_dialect.py

@@ -111,13 +111,18 @@ def test_does_not_access_dbapi_connection_directly(self) -> None:
                     "should use connection.exec_driver_sql() instead"
                 )
 
-    def test_returns_fallback_on_error(self) -> None:
-        """Should return (3, 0, 0) if the query fails."""
+    def test_returns_fallback_on_operational_error(self) -> None:
+        """Should return (3, 0, 0) if the server query fails with a
+        connection-level error. Unrelated errors (bugs) propagate."""
         from unittest.mock import MagicMock
 
+        import dqlitedbapi.exceptions
+
         dialect = DqliteDialect()
         mock_conn = MagicMock()
-        mock_conn.exec_driver_sql.side_effect = Exception("connection broken")
+        mock_conn.exec_driver_sql.side_effect = dqlitedbapi.exceptions.OperationalError(
+            "connection broken"
+        )
 
         result = dialect._get_server_version_info(mock_conn)
         assert result == (3, 0, 0)
@@ -300,23 +305,30 @@ def test_ping_returns_true_on_success(self) -> None:
         mock_conn = MagicMock()
         assert dialect.do_ping(mock_conn) is True
 
-    def test_ping_returns_false_on_error(self) -> None:
-        """do_ping should return False when query fails."""
+    def test_ping_returns_false_on_connection_error(self) -> None:
+        """do_ping returns False on connection-level errors."""
         from unittest.mock import MagicMock
 
+        import dqlitedbapi.exceptions
+
         dialect = DqliteDialect()
         mock_conn = MagicMock()
-        mock_conn.cursor.return_value.execute.side_effect = RuntimeError("boom")
+        mock_conn.cursor.return_value.execute.side_effect = (
+            dqlitedbapi.exceptions.OperationalError("bye")
+        )
         assert dialect.do_ping(mock_conn) is False
 
     def test_ping_closes_cursor_even_on_error(self) -> None:
-        """do_ping must close cursor even when execute fails."""
+        """do_ping must close cursor even when execute fails with a
+        connection-level error."""
         from unittest.mock import MagicMock
 
+        import dqlitedbapi.exceptions
+
         dialect = DqliteDialect()
         mock_conn = MagicMock()
         mock_cursor = MagicMock()
-        mock_cursor.execute.side_effect = RuntimeError("boom")
+        mock_cursor.execute.side_effect = dqlitedbapi.exceptions.OperationalError("bye")
         mock_conn.cursor.return_value = mock_cursor
 
         dialect.do_ping(mock_conn)

tests/test_dialect_dialect_config.py

@@ -0,0 +1,122 @@
+"""Dialect configuration tests covering recent hardening.
+
+- ISSUE-16 ``do_ping`` narrowed to connection-level exceptions only.
+- ISSUE-18 ``is_disconnect`` type-dispatches before falling back to
+  message substring matching.
+- ISSUE-19 ``create_connect_args`` plumbs the ``timeout`` URL query
+  through and rejects typos.
+- ISSUE-21 ``set_isolation_level`` explicitly rejects AUTOCOMMIT.
+- ISSUE-23 ``supported_isolation_levels`` is declared.
+"""
+
+from unittest.mock import MagicMock
+
+import dqliteclient.exceptions
+import dqlitedbapi.exceptions
+import pytest
+from sqlalchemy.engine import make_url
+from sqlalchemy.exc import ArgumentError
+
+from sqlalchemydqlite.base import DqliteDialect
+
+
+class TestSupportedIsolationLevels:
+    def test_declared(self) -> None:
+        assert DqliteDialect.supported_isolation_levels == ("SERIALIZABLE",)
+
+
+class TestSetIsolationLevel:
+    def test_serializable_is_noop(self) -> None:
+        dialect = DqliteDialect()
+        dialect.set_isolation_level(MagicMock(), "SERIALIZABLE")  # no raise
+
+    def test_autocommit_rejected(self) -> None:
+        dialect = DqliteDialect()
+        with pytest.raises(ArgumentError, match="AUTOCOMMIT"):
+            dialect.set_isolation_level(MagicMock(), "AUTOCOMMIT")
+
+    def test_other_levels_warn(self) -> None:
+        dialect = DqliteDialect()
+        with pytest.warns(UserWarning, match="SERIALIZABLE"):
+            dialect.set_isolation_level(MagicMock(), "READ COMMITTED")
+
+
+class TestCreateConnectArgsURLQuery:
+    def test_timeout_forwarded(self) -> None:
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db?timeout=5.5")
+        args, kwargs = dialect.create_connect_args(url)
+        assert kwargs["timeout"] == 5.5
+        assert kwargs["address"] == "host:19001"
+        assert kwargs["database"] == "db"
+
+    def test_no_query_params_still_works(self) -> None:
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db")
+        _, kwargs = dialect.create_connect_args(url)
+        assert "timeout" not in kwargs
+
+    def test_unknown_param_raises(self) -> None:
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db?timeoutt=5")  # typo
+        with pytest.raises(ArgumentError, match="timeoutt"):
+            dialect.create_connect_args(url)
+
+    def test_unparseable_timeout_raises(self) -> None:
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db?timeout=not-a-number")
+        with pytest.raises(ArgumentError, match="float"):
+            dialect.create_connect_args(url)
+
+
+class TestDoPingNarrowExceptions:
+    def test_returns_true_on_success(self) -> None:
+        dialect = DqliteDialect()
+        conn = MagicMock()
+        cursor = MagicMock()
+        conn.cursor.return_value = cursor
+        cursor.execute.return_value = None
+        assert dialect.do_ping(conn) is True
+
+    def test_returns_false_on_operational_error(self) -> None:
+        dialect = DqliteDialect()
+        conn = MagicMock()
+        cursor = MagicMock()
+        conn.cursor.return_value = cursor
+        cursor.execute.side_effect = dqlitedbapi.exceptions.OperationalError("bye")
+        assert dialect.do_ping(conn) is False
+
+    def test_propagates_unexpected_exception(self) -> None:
+        dialect = DqliteDialect()
+        conn = MagicMock()
+        cursor = MagicMock()
+        conn.cursor.return_value = cursor
+        cursor.execute.side_effect = RuntimeError("bug")
+        with pytest.raises(RuntimeError, match="bug"):
+            dialect.do_ping(conn)
+
+
+class TestIsDisconnectTypeDispatch:
+    def test_dqlite_connection_error_is_disconnect(self) -> None:
+        dialect = DqliteDialect()
+        e = dqliteclient.exceptions.DqliteConnectionError("boom")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_os_error_is_disconnect(self) -> None:
+        dialect = DqliteDialect()
+        e = OSError(111, "Connection refused")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_broken_pipe_is_disconnect(self) -> None:
+        dialect = DqliteDialect()
+        assert dialect.is_disconnect(BrokenPipeError(), None, None) is True
+
+    def test_leader_change_code_is_disconnect(self) -> None:
+        dialect = DqliteDialect()
+        e = dqliteclient.exceptions.OperationalError(10250, "not leader")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_unrelated_operational_error_is_not_disconnect(self) -> None:
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("no such table")
+        assert dialect.is_disconnect(e, None, None) is False