Extract _build_and_connect shared by sync and async connections

The construct-then-connect-then-wrap sequence lived twice — in
Connection._get_async_connection and AsyncConnection._ensure_connection
with the same kwargs and the same "Failed to connect: ..." wrapping.
A future governor addition would have to be threaded through both
sites or drift silently. Factor the body into a module-local
_build_and_connect coroutine in dqlitedbapi.connection and call it
from both flavours. Preserve the OperationalError message phrasing
verbatim.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -11,7 +11,7 @@
 from dqliteclient import DqliteConnection
 from dqliteclient.protocol import _validate_positive_int_or_none
 from dqlitedbapi.aio.cursor import AsyncCursor
-from dqlitedbapi.connection import _is_no_transaction_error
+from dqlitedbapi.connection import _build_and_connect, _is_no_transaction_error
 from dqlitedbapi.exceptions import InterfaceError, OperationalError, ProgrammingError
 
 __all__ = ["AsyncConnection"]
@@ -114,20 +114,14 @@ async def _ensure_connection(self) -> DqliteConnection:
             if self._async_conn is not None:
                 return self._async_conn
 
-            conn = DqliteConnection(
+            self._async_conn = await _build_and_connect(
                 self._address,
                 database=self._database,
                 timeout=self._timeout,
                 max_total_rows=self._max_total_rows,
                 max_continuation_frames=self._max_continuation_frames,
                 trust_server_heartbeat=self._trust_server_heartbeat,
             )
-            try:
-                await conn.connect()
-            except Exception as e:
-                raise OperationalError(f"Failed to connect: {e}") from e
-
-            self._async_conn = conn
 
         return self._async_conn
 

src/dqlitedbapi/connection.py

@@ -29,6 +29,38 @@
 _NO_TX_SUBSTRING = "no transaction is active"
 
 
+async def _build_and_connect(
+    address: str,
+    *,
+    database: str,
+    timeout: float,
+    max_total_rows: int | None,
+    max_continuation_frames: int | None,
+    trust_server_heartbeat: bool,
+) -> DqliteConnection:
+    """Build a DqliteConnection with the given governors and connect it.
+
+    Wraps the construct-then-connect sequence that both the sync and
+    async Connection flavours execute under their respective locks. The
+    ``OperationalError`` message phrasing ("Failed to connect: ...") is
+    intentionally verbatim so test assertions that match on the prefix
+    continue to pass.
+    """
+    conn = DqliteConnection(
+        address,
+        database=database,
+        timeout=timeout,
+        max_total_rows=max_total_rows,
+        max_continuation_frames=max_continuation_frames,
+        trust_server_heartbeat=trust_server_heartbeat,
+    )
+    try:
+        await conn.connect()
+    except Exception as e:
+        raise OperationalError(f"Failed to connect: {e}") from e
+    return conn
+
+
 def _is_no_transaction_error(exc: Exception) -> bool:
     """True if ``exc`` is a genuine "no active transaction" server reply.
 
@@ -254,19 +286,14 @@ async def _get_async_connection(self) -> DqliteConnection:
             if self._async_conn is not None:
                 return self._async_conn
 
-            conn = DqliteConnection(
+            self._async_conn = await _build_and_connect(
                 self._address,
                 database=self._database,
                 timeout=self._timeout,
                 max_total_rows=self._max_total_rows,
                 max_continuation_frames=self._max_continuation_frames,
                 trust_server_heartbeat=self._trust_server_heartbeat,
             )
-            try:
-                await conn.connect()
-            except Exception as e:
-                raise OperationalError(f"Failed to connect: {e}") from e
-            self._async_conn = conn
 
         return self._async_conn
 

tests/test_async_close_race.py

@@ -31,7 +31,7 @@ async def slow_execute(_sql: str, _params: object) -> tuple[int, int]:
     async def fake_query_raw_typed(_sql: str, _params: object) -> tuple[list, list, list]:
         return ([], [], [])
 
-    with patch("dqlitedbapi.aio.connection.DqliteConnection") as MockDqliteConn:
+    with patch("dqlitedbapi.connection.DqliteConnection") as MockDqliteConn:
         mock_instance = AsyncMock()
         mock_instance.connect = AsyncMock()
         mock_instance.execute = slow_execute

tests/test_async_connection_race.py

@@ -22,7 +22,7 @@ async def slow_connect() -> None:
             await asyncio.sleep(0.1)  # Simulate slow TCP handshake
             connect_finished.set()
 
-        with patch("dqlitedbapi.aio.connection.DqliteConnection") as MockDqliteConn:
+        with patch("dqlitedbapi.connection.DqliteConnection") as MockDqliteConn:
             mock_instance = AsyncMock()
             mock_instance.connect = slow_connect
             mock_instance._protocol = AsyncMock()

tests/test_protocol_serialization.py

@@ -42,7 +42,7 @@ async def mock_execute(sql: str, params: object) -> tuple:
             call_log.append((sql, "end"))
             return (0, 1)
 
-        with patch("dqlitedbapi.aio.connection.DqliteConnection") as MockDqliteConn:
+        with patch("dqlitedbapi.connection.DqliteConnection") as MockDqliteConn:
             mock_instance = AsyncMock()
             mock_instance.connect = AsyncMock()
             mock_instance.query_raw_typed = mock_query_raw_typed