Wrap client exceptions into PEP 249 hierarchy; serialize close

- cursor.execute used to let dqliteclient.exceptions.* leak through
  to the caller. PEP 249 requires all error classes to live under
  the module's own hierarchy. New _call_client helper wraps the
  client call, mapping:
    client.OperationalError     → dbapi.OperationalError
    client.DqliteConnectionError → dbapi.OperationalError
    client.ClusterError          → dbapi.OperationalError
    client.ProtocolError         → dbapi.InterfaceError
    client.DataError             → dbapi.DataError
    client.InterfaceError        → dbapi.InterfaceError
  Original cause preserved via `from`. Applied at both sync and
  async cursor entry points.

- AsyncConnection.close() now acquires _op_lock before tearing
  down the underlying protocol. A concurrent task mid-execute used
  to find the socket closed underneath it; close now waits for
  the in-flight operation to complete.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -79,13 +79,26 @@ async def connect(self) -> None:
         await self._ensure_connection()
 
     async def close(self) -> None:
-        """Close the connection."""
+        """Close the connection.
+
+        Serializes with any in-flight operation via ``_op_lock`` so we
+        never tear down the underlying protocol while another task is
+        mid-execute/mid-commit — that races would leave the caller
+        with mysterious "connection closed" errors mid-query.
+        """
         if self._closed:
             return
+        # Set _closed first so any task waiting on the lock sees the
+        # closed state as soon as it acquires. Then drain the current
+        # in-flight op (if any) under the lock.
         self._closed = True
-        if self._async_conn is not None:
-            await self._async_conn.close()
-            self._async_conn = None
+        if self._async_conn is None:
+            return
+        _, op_lock = self._ensure_locks()
+        async with op_lock:
+            if self._async_conn is not None:
+                await self._async_conn.close()
+                self._async_conn = None
 
     async def commit(self) -> None:
         """Commit any pending transaction.

src/dqlitedbapi/aio/cursor.py

@@ -3,7 +3,12 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
-from dqlitedbapi.cursor import _convert_params, _convert_row, _strip_leading_comments
+from dqlitedbapi.cursor import (
+    _call_client,
+    _convert_params,
+    _convert_row,
+    _strip_leading_comments,
+)
 from dqlitedbapi.exceptions import InterfaceError
 
 if TYPE_CHECKING:
@@ -82,7 +87,9 @@ async def execute(
         _, op_lock = self._connection._ensure_locks()
         async with op_lock:
             if is_query:
-                columns, column_types, rows = await conn.query_raw_typed(operation, params)
+                columns, column_types, rows = await _call_client(
+                    conn.query_raw_typed(operation, params)
+                )
                 self._description = [
                     (
                         name,
@@ -99,7 +106,7 @@ async def execute(
                 self._row_index = 0
                 self._rowcount = len(rows)
             else:
-                last_id, affected = await conn.execute(operation, params)
+                last_id, affected = await _call_client(conn.execute(operation, params))
                 self._lastrowid = last_id
                 self._rowcount = affected
                 self._description = None

src/dqlitedbapi/cursor.py

@@ -1,17 +1,54 @@
 """PEP 249 Cursor implementation for dqlite."""
 
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Coroutine, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+import dqliteclient.exceptions as _client_exc
 from dqlitewire.constants import ValueType
 
-from dqlitedbapi.exceptions import InterfaceError, ProgrammingError
+from dqlitedbapi.exceptions import (
+    DataError,
+    InterfaceError,
+    OperationalError,
+    ProgrammingError,
+)
+from dqlitedbapi.exceptions import (
+    InterfaceError as _DbapiInterfaceError,
+)
 from dqlitedbapi.types import (
     _convert_bind_param,
     _datetime_from_iso8601,
     _datetime_from_unixtime,
 )
 
+
+async def _call_client(coro: Coroutine[Any, Any, Any]) -> Any:
+    """Await a client-layer coroutine, mapping its exceptions into the
+    PEP 249 hierarchy. Preserves the original via ``from``.
+
+    Mapping:
+      client.OperationalError      → dbapi.OperationalError (same code/msg)
+      client.DqliteConnectionError → dbapi.OperationalError (network flavor)
+      client.ClusterError          → dbapi.OperationalError
+      client.ProtocolError         → dbapi.InterfaceError
+      client.DataError             → dbapi.DataError
+      client.InterfaceError        → dbapi.InterfaceError
+    """
+    try:
+        return await coro
+    except _client_exc.OperationalError as e:
+        raise OperationalError(str(e)) from e
+    except _client_exc.DqliteConnectionError as e:
+        raise OperationalError(str(e)) from e
+    except _client_exc.ClusterError as e:
+        raise OperationalError(str(e)) from e
+    except _client_exc.ProtocolError as e:
+        raise _DbapiInterfaceError(str(e)) from e
+    except _client_exc.DataError as e:
+        raise DataError(str(e)) from e
+    except _client_exc.InterfaceError as e:
+        raise _DbapiInterfaceError(str(e)) from e
+
 if TYPE_CHECKING:
     from dqlitedbapi.connection import Connection
 
@@ -165,7 +202,9 @@ async def _execute_async(self, operation: str, parameters: Sequence[Any] | None
         )
 
         if is_query:
-            columns, column_types, rows = await conn.query_raw_typed(operation, params)
+            columns, column_types, rows = await _call_client(
+                conn.query_raw_typed(operation, params)
+            )
             self._description = [
                 (
                     name,
@@ -182,7 +221,7 @@ async def _execute_async(self, operation: str, parameters: Sequence[Any] | None
             self._row_index = 0
             self._rowcount = len(rows)
         else:
-            last_id, affected = await conn.execute(operation, params)
+            last_id, affected = await _call_client(conn.execute(operation, params))
             self._lastrowid = last_id
             self._rowcount = affected
             self._description = None

tests/test_async_close_race.py

@@ -0,0 +1,65 @@
+"""AsyncConnection.close serializes with in-flight operations (ISSUE-32).
+
+Previously close() called await self._async_conn.close() without
+acquiring _op_lock; a concurrent task mid-execute would find the
+protocol torn down underneath it. Now close() acquires the lock so
+in-flight operations drain cleanly before the socket is closed.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from dqlitedbapi.aio.connection import AsyncConnection
+
+
+@pytest.mark.asyncio
+async def test_close_waits_for_in_flight_execute() -> None:
+    """If task A is mid-execute (holding _op_lock), task B's close
+    awaits that task before tearing down the protocol."""
+    conn = AsyncConnection("localhost:19001", database="x")
+
+    order: list[str] = []
+
+    async def slow_execute(_sql: str, _params: object) -> tuple[int, int]:
+        order.append("execute:start")
+        await asyncio.sleep(0.05)
+        order.append("execute:end")
+        return (0, 0)
+
+    async def fake_query_raw_typed(_sql: str, _params: object) -> tuple[list, list, list]:
+        return ([], [], [])
+
+    with patch("dqlitedbapi.aio.connection.DqliteConnection") as MockDqliteConn:
+        mock_instance = AsyncMock()
+        mock_instance.connect = AsyncMock()
+        mock_instance.execute = slow_execute
+        mock_instance.query_raw_typed = fake_query_raw_typed
+
+        async def fake_close() -> None:
+            order.append("close:start")
+            order.append("close:end")
+
+        mock_instance.close = fake_close
+        MockDqliteConn.return_value = mock_instance
+
+        await conn.connect()
+
+        async def run_execute() -> None:
+            cursor = conn.cursor()
+            await cursor.execute("INSERT INTO t VALUES (1)")
+
+        async def run_close() -> None:
+            await asyncio.sleep(0.01)  # let execute start first
+            await conn.close()
+
+        await asyncio.gather(run_execute(), run_close())
+
+    # execute must complete before close starts tearing down.
+    assert order == [
+        "execute:start",
+        "execute:end",
+        "close:start",
+        "close:end",
+    ], f"actual order: {order}"

tests/test_exception_mapping.py

@@ -0,0 +1,70 @@
+"""PEP 249 exception wrapping at the cursor layer (ISSUE-31).
+
+The underlying client raises ``dqliteclient.exceptions.*``. PEP 249
+requires a specific exception hierarchy under the DBAPI module, so the
+cursor must translate those to ``dqlitedbapi.exceptions.*`` when they
+surface to the user.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import dqliteclient.exceptions as client_exc
+import pytest
+
+import dqlitedbapi.exceptions as dbapi_exc
+from dqlitedbapi.cursor import Cursor
+
+
+def _cursor_with_async_conn_raising(exc: Exception) -> Cursor:
+    """Build a Cursor whose underlying async conn raises ``exc`` on every
+    query/execute. Bypasses the event-loop thread by running the coroutine
+    in a fresh event loop."""
+    mock_async_conn = AsyncMock()
+    mock_async_conn.query_raw_typed = AsyncMock(side_effect=exc)
+    mock_async_conn.execute = AsyncMock(side_effect=exc)
+
+    mock_conn = MagicMock()
+
+    async def get_async_conn() -> AsyncMock:
+        return mock_async_conn
+
+    mock_conn._get_async_connection = get_async_conn
+
+    def run_sync(coro: object) -> object:
+        return asyncio.new_event_loop().run_until_complete(coro)  # type: ignore[arg-type]
+
+    mock_conn._run_sync = run_sync
+    return Cursor(mock_conn)
+
+
+class TestExceptionWrapping:
+    def test_client_operational_error_becomes_dbapi_operational_error(self) -> None:
+        c = _cursor_with_async_conn_raising(client_exc.OperationalError(1, "boom"))
+        with pytest.raises(dbapi_exc.OperationalError, match="boom"):
+            c.execute("SELECT 1")
+
+    def test_client_connection_error_becomes_operational_error(self) -> None:
+        c = _cursor_with_async_conn_raising(client_exc.DqliteConnectionError("no route"))
+        with pytest.raises(dbapi_exc.OperationalError, match="no route"):
+            c.execute("SELECT 1")
+
+    def test_client_protocol_error_becomes_interface_error(self) -> None:
+        c = _cursor_with_async_conn_raising(client_exc.ProtocolError("bad frame"))
+        with pytest.raises(dbapi_exc.InterfaceError, match="bad frame"):
+            c.execute("SELECT 1")
+
+    def test_client_data_error_becomes_data_error(self) -> None:
+        c = _cursor_with_async_conn_raising(client_exc.DataError("bad param"))
+        with pytest.raises(dbapi_exc.DataError, match="bad param"):
+            c.execute("INSERT INTO t VALUES (?)", [object()])
+
+    def test_chained_cause_preserved(self) -> None:
+        original = client_exc.OperationalError(1, "original")
+        c = _cursor_with_async_conn_raising(original)
+        try:
+            c.execute("SELECT 1")
+        except dbapi_exc.OperationalError as e:
+            assert e.__cause__ is original
+        else:
+            pytest.fail("expected OperationalError")