fix(mypyc): isolate compiled async exception state

docs/changelog.rst

@@ -9,6 +9,14 @@ important operational fixes.
 Recent Updates
 ==============
 
+v0.61.1 - Compiled async exception handling
+------------------------------------------------------------------------------
+
+**Fixed:**
+
+* Compiled async drivers no longer re-raise an exception already being handled
+  by the caller when a database operation inside that handler succeeds.
+
 v0.61.0 - Scoped memory recall and ADK modernization
 ------------------------------------------------------------------------------
 

pyproject.toml

@@ -24,7 +24,7 @@ maintainers = [{ name = "Litestar Developers", email = "hello@litestar.dev" }]
 name = "sqlspec"
 readme = "README.md"
 requires-python = ">=3.10, <4.0"
-version = "0.61.0"
+version = "0.61.1"
 
 [project.urls]
 Discord = "https://discord.gg/litestar"
@@ -331,7 +331,7 @@ opt_level = "3"   # Maximum optimization (0-3)
 allow_dirty = true
 commit = false
 commit_args = "--no-verify"
-current_version = "0.61.0"
+current_version = "0.61.1"
 ignore_missing_files = false
 ignore_missing_version = false
 message = "chore(release): bump to v{new_version}"

sqlspec/adapters/aiomysql/core.py

@@ -190,21 +190,21 @@ def __init__(self, driver: Any, sql: str, parameters: Any, chunk_size: int, json
         self._row_plan: tuple[list[str], list[int] | None] | None = None
 
     async def start(self) -> None:
-        from aiomysql import SSCursor
-
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            cursor = await self._driver.connection.cursor(SSCursor)
-            self._cursor = cursor
-            await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
-            self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        from aiomysql import SSCursor
+
+        cursor = await self._driver.connection.cursor(SSCursor)
+        self._cursor = cursor
+        await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
+        self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        rows: list[Any] = []
-        async with handler:
-            rows = await self._cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []

sqlspec/adapters/aiosqlite/core.py

@@ -257,18 +257,18 @@ def __init__(self, driver: Any, sql: str, parameters: Any, chunk_size: int) -> N
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            cursor = await self._driver.connection.cursor()
-            cursor.arraysize = self._chunk_size
-            self._cursor = cursor
-            await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        cursor = await self._driver.connection.cursor()
+        cursor.arraysize = self._chunk_size
+        self._cursor = cursor
+        await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        rows: list[Any] = []
-        async with handler:
-            rows = await self._cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []

sqlspec/adapters/asyncmy/core.py

@@ -162,21 +162,21 @@ def __init__(self, driver: Any, sql: str, parameters: Any, chunk_size: int, json
         self._row_plan: tuple[list[str], list[int] | None] | None = None
 
     async def start(self) -> None:
-        from asyncmy.cursors import SSCursor
-
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            cursor = self._driver.connection.cursor(SSCursor)
-            self._cursor = cursor
-            await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
-            self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        from asyncmy.cursors import SSCursor
+
+        cursor = self._driver.connection.cursor(SSCursor)
+        self._cursor = cursor
+        await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
+        self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        rows: list[Any] = []
-        async with handler:
-            rows = await self._cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []

sqlspec/adapters/asyncpg/core.py

@@ -402,24 +402,25 @@ def __init__(self, driver: Any, sql: str, parameters: "tuple[Any, ...]", chunk_s
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            transaction = self._driver.connection.transaction()
-            await transaction.start()
-            self._transaction = transaction
-            try:
-                self._cursor = await self._driver.connection.cursor(self._sql, *self._parameters)
-            except BaseException:
-                await transaction.rollback()
-                self._transaction = None
-                raise
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        transaction = self._driver.connection.transaction()
+        await transaction.start()
+        self._transaction = transaction
+        try:
+            self._cursor = await self._driver.connection.cursor(self._sql, *self._parameters)
+        except BaseException:
+            await transaction.rollback()
+            self._transaction = None
+            raise
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        records: list[Any] = []
-        async with handler:
-            records = await self._cursor.fetch(self._chunk_size)
+        records = await self._driver._run_with_exception_handler(handler, self._cursor.fetch, self._chunk_size)
         self._driver._check_pending_exception(handler)
+        assert records is not None
         return [dict(record) for record in records]
 
     async def close(self, error: bool = False) -> None:

sqlspec/adapters/mysqlconnector/core.py

@@ -277,18 +277,18 @@ def __init__(
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            cursor = await self._driver.connection.cursor(**self._cursor_options)
-            self._cursor = cursor
-            await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
-            self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        cursor = await self._driver.connection.cursor(**self._cursor_options)
+        self._cursor = cursor
+        await cursor.execute(self._sql, normalize_execute_parameters(self._parameters))
+        self._row_plan = resolve_row_plan(self._cursor.description, self._json_type_codes)
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        rows: list[Any] = []
-        async with handler:
-            rows = await self._cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []

sqlspec/adapters/oracledb/core.py

@@ -43,7 +43,7 @@
 from sqlspec.utils.type_guards import has_rowcount, is_readable
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Mapping
+    from collections.abc import Awaitable, Callable, Mapping
 
     from sqlspec.adapters.oracledb._typing import (
         OracleAsyncConnection,
@@ -154,7 +154,7 @@ def _resolve_oracledb_version() -> "tuple[int, int, int]":
         version = oracledb.__version__
     except AttributeError:
         version = "0.0.0"
-    return _parse_version_tuple(version)
+    return _parse_version_tuple(str(version))
 
 
 ORACLEDB_VERSION: "tuple[int, int, int]" = _resolve_oracledb_version()
@@ -718,34 +718,34 @@ def __init__(
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            cursor = self._driver.connection.cursor()
-            self._cursor = cursor
-            cursor.arraysize = self._chunk_size
-            cursor.prefetchrows = self._chunk_size
-            fetch_kwargs = build_fetch_kwargs(self._driver.driver_features)
-            if self._fetch_lobs is not None:
-                fetch_kwargs["fetch_lobs"] = self._fetch_lobs
-            parameters = await coerce_large_parameters_async(
-                self._driver.connection,
-                self._parameters,
-                clob_type=DB_TYPE_CLOB,
-                blob_type=DB_TYPE_BLOB,
-                varchar2_byte_limit=self._driver.driver_features.get("oracle_varchar2_byte_limit", 4000),
-                raw_byte_limit=self._driver.driver_features.get("oracle_raw_byte_limit", 2000),
-                version_cache=getattr(self._driver, "_oracle_version_cache", None),
-            )
-            await cast("Any", cursor).execute(self._sql, parameters or {}, **fetch_kwargs)
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        cursor = self._driver.connection.cursor()
+        self._cursor = cursor
+        cursor.arraysize = self._chunk_size
+        cursor.prefetchrows = self._chunk_size
+        fetch_kwargs = build_fetch_kwargs(self._driver.driver_features)
+        if self._fetch_lobs is not None:
+            fetch_kwargs["fetch_lobs"] = self._fetch_lobs
+        parameters = await coerce_large_parameters_async(
+            self._driver.connection,
+            self._parameters,
+            clob_type=DB_TYPE_CLOB,
+            blob_type=DB_TYPE_BLOB,
+            varchar2_byte_limit=self._driver.driver_features.get("oracle_varchar2_byte_limit", 4000),
+            raw_byte_limit=self._driver.driver_features.get("oracle_raw_byte_limit", 2000),
+            version_cache=getattr(self._driver, "_oracle_version_cache", None),
+        )
+        await cast("Any", cursor).execute(self._sql, parameters or {}, **fetch_kwargs)
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
         cursor = self._cursor
         if cursor is None:
             return []
-        rows: list[Any] = []
-        async with handler:
-            rows = await cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []
@@ -952,6 +952,14 @@ def handle_database_exceptions(self) -> "AsyncExceptionHandler": ...
 
     def _check_pending_exception(self, exc_handler: "AsyncExceptionHandler") -> None: ...
 
+    async def _run_with_exception_handler(
+        self,
+        exc_handler: "AsyncExceptionHandler",
+        operation: "Callable[..., Awaitable[Any]]",
+        *args: Any,
+        **kwargs: Any,
+    ) -> Any: ...
+
     def _resolve_row_metadata(self, description: object) -> tuple[list[str], bool]: ...
 
 

sqlspec/adapters/psqlpy/core.py

@@ -219,26 +219,26 @@ def __init__(self, driver: Any, sql: str, parameters: Any, chunk_size: int) -> N
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            transaction = self._driver.connection.transaction()
-            await transaction.begin()
-            self._transaction = transaction
-            try:
-                cursor = transaction.cursor(self._sql, self._parameters, array_size=self._chunk_size)
-                await cursor.start()
-                self._cursor = cursor
-            except BaseException:
-                self._transaction = None
-                with contextlib.suppress(Exception):
-                    await transaction.rollback()
-                raise
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        transaction = self._driver.connection.transaction()
+        await transaction.begin()
+        self._transaction = transaction
+        try:
+            cursor = transaction.cursor(self._sql, self._parameters, array_size=self._chunk_size)
+            await cursor.start()
+            self._cursor = cursor
+        except BaseException:
+            self._transaction = None
+            with contextlib.suppress(Exception):
+                await transaction.rollback()
+            raise
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        query_result: Any = None
-        async with handler:
-            query_result = await self._cursor.fetchmany(self._chunk_size)
+        query_result = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if query_result is None:
             return []

sqlspec/adapters/psycopg/core.py

@@ -310,27 +310,27 @@ def __init__(self, driver: Any, sql: str, parameters: Any, chunk_size: int) -> N
 
     async def start(self) -> None:
         handler = self._driver.handle_database_exceptions()
-        async with handler:
-            transaction = self._driver.connection.transaction()
-            await transaction.__aenter__()
-            self._transaction = transaction
-            try:
-                cursor = self._driver.connection.cursor(name=f"sqlspec_stream_{uuid4().hex}")
-                cursor.itersize = self._chunk_size
-                await execute_with_optional_parameters_async(cursor, self._sql, self._parameters)
-                self._cursor = cursor
-            except BaseException as exc:
-                self._transaction = None
-                with contextlib.suppress(Exception):
-                    await transaction.__aexit__(type(exc), exc, exc.__traceback__)
-                raise
+        await self._driver._run_with_exception_handler(handler, self._start)
         self._driver._check_pending_exception(handler)
 
+    async def _start(self) -> None:
+        transaction = self._driver.connection.transaction()
+        await transaction.__aenter__()
+        self._transaction = transaction
+        try:
+            cursor = self._driver.connection.cursor(name=f"sqlspec_stream_{uuid4().hex}")
+            cursor.itersize = self._chunk_size
+            await execute_with_optional_parameters_async(cursor, self._sql, self._parameters)
+            self._cursor = cursor
+        except BaseException as exc:
+            self._transaction = None
+            with contextlib.suppress(Exception):
+                await transaction.__aexit__(type(exc), exc, exc.__traceback__)
+            raise
+
     async def fetch_chunk(self) -> "list[dict[str, Any]]":
         handler = self._driver.handle_database_exceptions()
-        rows: list[Any] = []
-        async with handler:
-            rows = await self._cursor.fetchmany(self._chunk_size)
+        rows = await self._driver._run_with_exception_handler(handler, self._cursor.fetchmany, self._chunk_size)
         self._driver._check_pending_exception(handler)
         if not rows:
             return []