Use public API in _get_server_version_info and add error handling

Replace internal .dbapi_connection access with connection.exec_driver_sql()
and wrap in try/except to gracefully fall back to (3, 0, 0) on failure.

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

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -107,14 +107,13 @@ def _get_server_version_info(self, connection: Any) -> tuple[int, ...]:
 
         dqlite uses SQLite internally, so we return SQLite version.
         """
-        cursor = connection.connection.dbapi_connection.cursor()
-        cursor.execute("SELECT sqlite_version()")
-        row = cursor.fetchone()
-        cursor.close()
-
-        if row:
-            version_str = row[0]
-            return tuple(int(x) for x in version_str.split("."))
+        try:
+            result = connection.exec_driver_sql("SELECT sqlite_version()")
+            version_str = result.scalar()
+            if version_str:
+                return tuple(int(x) for x in version_str.split("."))
+        except Exception:
+            pass
         return (3, 0, 0)
 
 

tests/test_dialect.py

@@ -93,6 +93,49 @@ def test_create_async_engine(self) -> None:
         assert engine.dialect.driver == "dqlitedbapi_aio"
 
 
+class TestGetServerVersionInfo:
+    def test_does_not_access_dbapi_connection_directly(self) -> None:
+        """_get_server_version_info should use exec_driver_sql, not internal attributes."""
+        import ast
+        import inspect
+        import textwrap
+
+        source = textwrap.dedent(inspect.getsource(DqliteDialect._get_server_version_info))
+        tree = ast.parse(source)
+
+        # Check that it doesn't access .dbapi_connection
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Attribute) and node.attr == "dbapi_connection":
+                raise AssertionError(
+                    "_get_server_version_info accesses .dbapi_connection directly; "
+                    "should use connection.exec_driver_sql() instead"
+                )
+
+    def test_returns_fallback_on_error(self) -> None:
+        """Should return (3, 0, 0) if the query fails."""
+        from unittest.mock import MagicMock
+
+        dialect = DqliteDialect()
+        mock_conn = MagicMock()
+        mock_conn.exec_driver_sql.side_effect = Exception("connection broken")
+
+        result = dialect._get_server_version_info(mock_conn)
+        assert result == (3, 0, 0)
+
+    def test_parses_version_string(self) -> None:
+        """Should parse a version string like '3.39.4' into a tuple."""
+        from unittest.mock import MagicMock
+
+        dialect = DqliteDialect()
+        mock_conn = MagicMock()
+        mock_result = MagicMock()
+        mock_result.scalar.return_value = "3.39.4"
+        mock_conn.exec_driver_sql.return_value = mock_result
+
+        result = dialect._get_server_version_info(mock_conn)
+        assert result == (3, 39, 4)
+
+
 class TestURLParsing:
     def test_parse_basic_url(self) -> None:
         url = URL.create("dqlite", host="localhost", port=9001, database="test")