Add is_disconnect() to detect broken dqlite connections

The inherited SQLiteDialect.is_disconnect() only checks for pysqlite
patterns. Add dqlite-specific detection for network errors like
"Connection closed", "timed out", "Failed to connect", etc. so the
pool can properly invalidate broken connections.

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

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -92,6 +92,29 @@ def do_commit(self, dbapi_connection: DBAPIConnection) -> None:
             if "no transaction is active" not in str(e):
                 raise
 
+    _dqlite_disconnect_messages = (
+        "Connection closed",
+        "timed out",
+        "Failed to connect",
+        "not connected",
+        "Not connected",
+    )
+
+    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.
+        """
+        import dqlitedbapi.exceptions
+
+        if isinstance(e, dqlitedbapi.exceptions.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."""
         try:

tests/test_dialect.py

@@ -152,6 +152,54 @@ def test_async_dialect_returns_underlying_connection(self) -> None:
         )
 
 
+class TestIsDisconnect:
+    def test_recognizes_connection_closed(self) -> None:
+        """is_disconnect should return True for connection-closed errors."""
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("Connection closed by server")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_recognizes_failed_to_connect(self) -> None:
+        """is_disconnect should return True for connection-failure errors."""
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("Failed to connect: refused")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_does_not_flag_normal_errors(self) -> None:
+        """is_disconnect should return False for normal operational errors."""
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("no such table: users")
+        assert dialect.is_disconnect(e, None, None) is False
+
+    def test_recognizes_not_connected(self) -> None:
+        """is_disconnect should return True for 'not connected' errors."""
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("Not connected to database")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_recognizes_timed_out(self) -> None:
+        """is_disconnect should return True for timeout errors."""
+        import dqlitedbapi.exceptions
+
+        dialect = DqliteDialect()
+        e = dqlitedbapi.exceptions.OperationalError("Connection timed out")
+        assert dialect.is_disconnect(e, None, None) is True
+
+    def test_is_defined_on_dialect(self) -> None:
+        """DqliteDialect must define its own is_disconnect, not just inherit."""
+        assert "is_disconnect" in DqliteDialect.__dict__, (
+            "DqliteDialect must override is_disconnect"
+        )
+
+
 class TestURLParsing:
     def test_parse_basic_url(self) -> None:
         url = URL.create("dqlite", host="localhost", port=9001, database="test")