diff --git a/mssql_python/async_query/async_connection.py b/mssql_python/async_query/async_connection.py index 327251b32..8f589285a 100644 --- a/mssql_python/async_query/async_connection.py +++ b/mssql_python/async_query/async_connection.py @@ -46,8 +46,8 @@ class AsyncConnection: ProgrammingError = ProgrammingError NotSupportedError = NotSupportedError - def __init__(self, native_connection: Any) -> None: - self._native_connection = native_connection + def __init__(self, py_core_async_connection: Any) -> None: + self._py_core_async_connection = py_core_async_connection @classmethod async def connect( @@ -58,54 +58,61 @@ async def connect( python_logger: Optional[Any] = None, ) -> "AsyncConnection": """Establish an asynchronous connection from an ODBC connection string.""" + logger_bridge = python_logger + if logger_bridge is None and logger.is_debug_enabled: + logger_bridge = logger logger.debug( - "AsyncConnection.connect: starting; autocommit=%s; custom_logger=%s", + "AsyncConnection.connect: starting; autocommit=%s; logger_source=%s", autocommit, - python_logger is not None, + ( + "custom" + if python_logger is not None + else "mssql_python" if logger_bridge is not None else "disabled" + ), ) with translate_py_core_exceptions(): client_context_dict = build_async_connection_context(connection_str, timeout) py_core = load_py_core() - native_connection = await py_core.PyAsyncConnection.connect( + py_core_async_connection = await py_core.PyAsyncConnection.connect( client_context_dict, - python_logger=python_logger, + python_logger=logger_bridge, autocommit=autocommit, ) logger.debug("AsyncConnection.connect: connected") - return cls(native_connection) + return cls(py_core_async_connection) def cursor(self) -> AsyncCursor: """Create a public asynchronous cursor sharing this connection.""" with translate_py_core_exceptions(): - native_cursor = self._native_connection.cursor() + py_core_async_cursor = self._py_core_async_connection.cursor() logger.debug("AsyncConnection.cursor: cursor created") - return AsyncCursor(native_cursor) + return AsyncCursor(py_core_async_cursor, self) async def commit(self) -> None: """Commit the active transaction, if any.""" logger.debug("AsyncConnection.commit: starting") with translate_py_core_exceptions(): - await self._native_connection.commit() + await self._py_core_async_connection.commit() logger.debug("AsyncConnection.commit: completed") async def rollback(self) -> None: """Roll back the active transaction, if any.""" logger.debug("AsyncConnection.rollback: starting") with translate_py_core_exceptions(): - await self._native_connection.rollback() + await self._py_core_async_connection.rollback() logger.debug("AsyncConnection.rollback: completed") async def close(self) -> None: - """Close the native connection.""" + """Close the py-core async connection.""" logger.debug("AsyncConnection.close: starting") with translate_py_core_exceptions(): - await self._native_connection.close() + await self._py_core_async_connection.close() logger.debug("AsyncConnection.close: completed") async def __aenter__(self) -> "AsyncConnection": logger.debug("AsyncConnection.__aenter__: entering context") with translate_py_core_exceptions(): - await self._native_connection.__aenter__() + await self._py_core_async_connection.__aenter__() logger.debug("AsyncConnection.__aenter__: context entered") return self @@ -115,7 +122,7 @@ async def __aexit__(self, exc_type, exc_value, traceback) -> Any: exc_type is not None, ) with translate_py_core_exceptions(): - result = await self._native_connection.__aexit__(exc_type, exc_value, traceback) + result = await self._py_core_async_connection.__aexit__(exc_type, exc_value, traceback) logger.debug("AsyncConnection.__aexit__: context exited") return result @@ -123,30 +130,30 @@ async def __aexit__(self, exc_type, exc_value, traceback) -> Any: def timeout(self) -> int: """Default query timeout inherited by subsequently created cursors.""" with translate_py_core_exceptions(): - return self._native_connection.timeout + return self._py_core_async_connection.timeout @timeout.setter def timeout(self, value: int) -> None: with translate_py_core_exceptions(): - self._native_connection.timeout = value + self._py_core_async_connection.timeout = value logger.debug("AsyncConnection.timeout: updated") @property def autocommit(self) -> bool: """Whether the connection was opened in autocommit mode.""" with translate_py_core_exceptions(): - return self._native_connection.autocommit + return self._py_core_async_connection.autocommit @property def closed(self) -> bool: - """Whether close has been initiated on the native connection.""" + """Whether close has been initiated on the py-core async connection.""" with translate_py_core_exceptions(): - return self._native_connection.closed + return self._py_core_async_connection.closed def is_connected(self) -> bool: - """Return whether the native connection remains open.""" + """Return whether the py-core async connection remains open.""" with translate_py_core_exceptions(): - return self._native_connection.is_connected() + return self._py_core_async_connection.is_connected() def __repr__(self) -> str: state = "closed" if self.closed else "connected" diff --git a/mssql_python/async_query/async_cursor.py b/mssql_python/async_query/async_cursor.py index 3e201c48a..cf56ea9b2 100644 --- a/mssql_python/async_query/async_cursor.py +++ b/mssql_python/async_query/async_cursor.py @@ -6,9 +6,14 @@ may change without notice. """ +from collections.abc import Mapping, Sequence from typing import Any, Optional +import uuid +from ..helpers import get_settings from ..logging import logger +from ..row import Row +from . import async_execute, async_fetch from .exception_translator import translate_py_core_exceptions @@ -20,8 +25,63 @@ class AsyncCursor: Its signatures, behavior, error handling, and compatibility may change without notice. """ - def __init__(self, native_cursor: Any) -> None: - self._native_cursor = native_cursor + def __init__(self, py_core_async_cursor: Any, connection: Any = None) -> None: + self._py_core_async_cursor = py_core_async_cursor + self._connection = connection + self._closed = False + self._fetched_row_count = 0 + self._fetch_rowcount: int | None = None + self._description: list[tuple[Any, ...]] | None = None + self._column_map: dict[str, int] = {} + self._column_map_lower: dict[str, int] | None = None + self._uuid_str_indices: tuple[int, ...] | None = None + + def _clear_result_metadata(self) -> None: + self._description = None + self._column_map = {} + self._column_map_lower = None + self._uuid_str_indices = None + + def _initialize_result_metadata(self) -> None: + with translate_py_core_exceptions(): + description = self._py_core_async_cursor.description + if description is None: + self._clear_result_metadata() + return + + settings = get_settings() + self._description = [ + ((column[0].lower() if settings.lowercase else column[0]), *column[1:]) + for column in description + ] + self._column_map = {column[0]: index for index, column in enumerate(self._description)} + self._column_map_lower = ( + {name.lower(): index for name, index in self._column_map.items()} + if settings.lowercase + else None + ) + self._uuid_str_indices = ( + tuple(index for index, column in enumerate(self._description) if column[1] is uuid.UUID) + if not settings.native_uuid + else None + ) + + def _reset_fetch_tracking(self) -> None: + self._fetched_row_count = 0 + self._fetch_rowcount = None + + def _check_closed(self) -> None: + if self._closed or (self._connection is not None and self._connection.closed): + message = "Cursor is closed" if self._closed else "Connection is closed" + with translate_py_core_exceptions(): + raise RuntimeError(message) + + def _record_fetch(self, count: int, exhausted: bool) -> None: + if count: + self._fetched_row_count += count + self._fetch_rowcount = self._fetched_row_count + elif exhausted and self._fetched_row_count == 0: + self._fetch_rowcount = 0 async def execute( self, @@ -30,86 +90,80 @@ async def execute( use_prepare: bool = True, reset_cursor: bool = True, ) -> "AsyncCursor": - if len(parameters) == 1 and isinstance(parameters[0], (tuple, list)): - parameters = tuple(parameters[0]) - - logger.debug("AsyncCursor.execute: starting") - with translate_py_core_exceptions(): - await self._native_cursor.execute( - operation, - *parameters, - use_prepare=use_prepare, - reset_cursor=reset_cursor, - ) - logger.debug("AsyncCursor.execute: completed") - return self + return await async_execute.execute( + self, + operation, + *parameters, + use_prepare=use_prepare, + reset_cursor=reset_cursor, + ) async def executemany( self, operation: str, - seq_of_parameters: Any, + seq_of_parameters: Sequence[Sequence[Any]] | Sequence[Mapping[str, Any]], *, use_prepare: bool = True, - ) -> "AsyncCursor": - logger.debug("AsyncCursor.executemany: starting") - with translate_py_core_exceptions(): - await self._native_cursor.executemany( - operation, - seq_of_parameters, - use_prepare=use_prepare, - ) - logger.debug("AsyncCursor.executemany: completed") - return self - - async def fetchone(self) -> Any: - with translate_py_core_exceptions(): - return await self._native_cursor.fetchone() + ) -> None: + await async_execute.executemany( + self, + operation, + seq_of_parameters, + use_prepare=use_prepare, + ) - async def fetchmany(self, size: Optional[int] = None) -> Any: - with translate_py_core_exceptions(): - if size is None: - return await self._native_cursor.fetchmany() - return await self._native_cursor.fetchmany(size) + async def fetchone(self) -> Row | None: + return await async_fetch.fetchone(self) - async def fetchall(self) -> Any: - with translate_py_core_exceptions(): - return await self._native_cursor.fetchall() + async def fetchmany(self, size: Optional[int] = None) -> list[Row]: + return await async_fetch.fetchmany(self, size) + + async def fetchall(self) -> list[Row]: + return await async_fetch.fetchall(self) async def nextset(self) -> bool: + self._reset_fetch_tracking() + self._clear_result_metadata() with translate_py_core_exceptions(): - return await self._native_cursor.nextset() + has_next = await self._py_core_async_cursor.nextset() + if has_next: + self._initialize_result_metadata() + return has_next async def close(self) -> None: logger.debug("AsyncCursor.close: starting") with translate_py_core_exceptions(): - await self._native_cursor.close() + await self._py_core_async_cursor.close() + self._closed = True + self._reset_fetch_tracking() logger.debug("AsyncCursor.close: completed") def setinputsizes(self, sizes: Any) -> None: with translate_py_core_exceptions(): - self._native_cursor.setinputsizes(sizes) + self._py_core_async_cursor.setinputsizes(sizes) @property def timeout(self) -> int: with translate_py_core_exceptions(): - return self._native_cursor.timeout + return self._py_core_async_cursor.timeout @property def description(self) -> Any: - with translate_py_core_exceptions(): - return self._native_cursor.description + return self._description @property def rowcount(self) -> int: + if self._fetch_rowcount is not None: + return self._fetch_rowcount with translate_py_core_exceptions(): - return self._native_cursor.rowcount + return self._py_core_async_cursor.rowcount @property def arraysize(self) -> int: with translate_py_core_exceptions(): - return self._native_cursor.arraysize + return self._py_core_async_cursor.arraysize @arraysize.setter def arraysize(self, value: int) -> None: with translate_py_core_exceptions(): - self._native_cursor.arraysize = value + self._py_core_async_cursor.arraysize = value diff --git a/mssql_python/async_query/async_execute.py b/mssql_python/async_query/async_execute.py new file mode 100644 index 000000000..27a5905c4 --- /dev/null +++ b/mssql_python/async_query/async_execute.py @@ -0,0 +1,79 @@ +"""Asynchronous statement execution through mssql-py-core.""" + +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING, Any + +from ..logging import logger +from ..row import Row +from .exception_translator import translate_py_core_exceptions + +if TYPE_CHECKING: + from .async_cursor import AsyncCursor + + +def _get_py_core_async_cursor(cursor: "AsyncCursor") -> Any: + return cursor._py_core_async_cursor # pyright: ignore[reportPrivateUsage] + + +async def execute( + cursor: "AsyncCursor", + operation: str, + *parameters: Any, + use_prepare: bool = True, + reset_cursor: bool = True, +) -> "AsyncCursor": + """Execute a statement using the py-core async cursor.""" + cursor._reset_fetch_tracking() # pyright: ignore[reportPrivateUsage] + cursor._clear_result_metadata() # pyright: ignore[reportPrivateUsage] + if len(parameters) == 1 and isinstance(parameters[0], (tuple, list, Row)): + parameters = tuple(parameters[0]) + + logger.debug( + "AsyncCursor.execute: starting; param_count=%d; use_prepare=%s; reset_cursor=%s", + len(parameters), + use_prepare, + reset_cursor, + ) + with translate_py_core_exceptions(): + await _get_py_core_async_cursor(cursor).execute( + operation, + *parameters, + use_prepare=use_prepare, + reset_cursor=reset_cursor, + ) + cursor._initialize_result_metadata() # pyright: ignore[reportPrivateUsage] + description = cursor.description + logger.debug( + "AsyncCursor.execute: completed; rowcount=%d; column_count=%d; has_result_set=%s", + cursor.rowcount, + len(description) if description is not None else 0, + description is not None, + ) + return cursor + + +async def executemany( + cursor: "AsyncCursor", + operation: str, + seq_of_parameters: Sequence[Sequence[Any]] | Sequence[Mapping[str, Any]], + *, + use_prepare: bool = True, +) -> None: + """Execute a statement for every parameter row using the py-core async cursor.""" + cursor._check_closed() # pyright: ignore[reportPrivateUsage] + batch_count = len(seq_of_parameters) + cursor._reset_fetch_tracking() # pyright: ignore[reportPrivateUsage] + cursor._clear_result_metadata() # pyright: ignore[reportPrivateUsage] + logger.debug( + "AsyncCursor.executemany: starting; batch_count=%d; use_prepare=%s", + batch_count, + use_prepare, + ) + with translate_py_core_exceptions(): + await _get_py_core_async_cursor(cursor).executemany( + operation, + seq_of_parameters, + use_prepare=use_prepare, + ) + cursor._initialize_result_metadata() # pyright: ignore[reportPrivateUsage] + logger.debug("AsyncCursor.executemany: completed; rowcount=%d", cursor.rowcount) diff --git a/mssql_python/async_query/async_fetch.py b/mssql_python/async_query/async_fetch.py new file mode 100644 index 000000000..e6f75e409 --- /dev/null +++ b/mssql_python/async_query/async_fetch.py @@ -0,0 +1,73 @@ +"""Asynchronous result fetching through mssql-py-core.""" + +from typing import TYPE_CHECKING, Any + +from ..logging import logger +from ..row import Row +from .exception_translator import translate_py_core_exceptions + +if TYPE_CHECKING: + from .async_cursor import AsyncCursor + + +def _get_py_core_async_cursor(cursor: "AsyncCursor") -> Any: + return cursor._py_core_async_cursor # pyright: ignore[reportPrivateUsage] + + +def _wrap_row(cursor: "AsyncCursor", values: tuple[Any, ...]) -> Row: + return Row( + values, + cursor._column_map, # pyright: ignore[reportPrivateUsage] + uuid_str_indices=cursor._uuid_str_indices, # pyright: ignore[reportPrivateUsage] + column_map_lower=cursor._column_map_lower, # pyright: ignore[reportPrivateUsage] + ) + + +async def fetchone(cursor: "AsyncCursor") -> Row | None: + """Fetch the next row through the py-core async cursor.""" + logger.debug("AsyncCursor.fetchone: starting") + with translate_py_core_exceptions(): + row = await _get_py_core_async_cursor(cursor).fetchone() + cursor._record_fetch(row is not None, row is None) # pyright: ignore[reportPrivateUsage] + logger.debug( + "AsyncCursor.fetchone: completed; row_found=%s; rowcount=%d", + row is not None, + cursor.rowcount, + ) + return None if row is None else _wrap_row(cursor, row) + + +async def fetchmany(cursor: "AsyncCursor", size: int | None = None) -> list[Row]: + """Fetch up to size rows, using cursor arraysize when size is omitted.""" + cursor._check_closed() # pyright: ignore[reportPrivateUsage] + requested_size = cursor.arraysize if size is None else size + logger.debug("AsyncCursor.fetchmany: starting; requested_size=%s", requested_size) + if requested_size <= 0: + logger.debug("AsyncCursor.fetchmany: completed; row_count=0; rowcount=%d", cursor.rowcount) + return [] + with translate_py_core_exceptions(): + if size is None: + rows = await _get_py_core_async_cursor(cursor).fetchmany() + else: + rows = await _get_py_core_async_cursor(cursor).fetchmany(size) + cursor._record_fetch(len(rows), not rows) # pyright: ignore[reportPrivateUsage] + logger.debug( + "AsyncCursor.fetchmany: completed; row_count=%d; rowcount=%d", + len(rows), + cursor.rowcount, + ) + return [_wrap_row(cursor, row) for row in rows] + + +async def fetchall(cursor: "AsyncCursor") -> list[Row]: + """Fetch all remaining rows through the py-core async cursor.""" + logger.debug("AsyncCursor.fetchall: starting") + with translate_py_core_exceptions(): + rows = await _get_py_core_async_cursor(cursor).fetchall() + cursor._record_fetch(len(rows), not rows) # pyright: ignore[reportPrivateUsage] + logger.debug( + "AsyncCursor.fetchall: completed; row_count=%d; rowcount=%d", + len(rows), + cursor.rowcount, + ) + return [_wrap_row(cursor, row) for row in rows] diff --git a/mssql_python/async_query/exception_translator.py b/mssql_python/async_query/exception_translator.py index fe0c72cfb..ae82f99e3 100644 --- a/mssql_python/async_query/exception_translator.py +++ b/mssql_python/async_query/exception_translator.py @@ -32,15 +32,61 @@ _ASYNC_DRIVER_ERROR = "Async operation failed" +_PROGRAMMING_RUNTIME_ERRORS = ("Cursor is closed",) +_INTERFACE_RUNTIME_ERRORS = ( + "Connection is closing", + "Connection is closed", +) +_OPERATIONAL_RUNTIME_ERROR_PREFIXES = ( + "Connection is broken", + "Connection is busy", +) +_PROGRAMMING_TYPE_ERROR_PREFIXES = ( + "The SQL contains ", + "Parameter style mismatch:", + "Named parameter cannot be empty", +) +_DATA_ERROR_NUMBERS = {241, 245, 248, 8114, 8115, 8134, 8152, 2628} +_INTEGRITY_ERROR_NUMBERS = {515, 547, 2601, 2627} +_PROGRAMMING_ERROR_NUMBERS = {102, 156, 201, 207, 208, 2812, 8144} + + +def _translate_known_builtin_error(error: Exception) -> Exception: + message = str(error) + if isinstance(error, RuntimeError): + if message in _PROGRAMMING_RUNTIME_ERRORS: + return ProgrammingError(_ASYNC_DRIVER_ERROR, message) + if message in _INTERFACE_RUNTIME_ERRORS: + return InterfaceError(_ASYNC_DRIVER_ERROR, message) + if message.startswith(_OPERATIONAL_RUNTIME_ERROR_PREFIXES): + return OperationalError(_ASYNC_DRIVER_ERROR, message) + if isinstance(error, TypeError) and message.startswith(_PROGRAMMING_TYPE_ERROR_PREFIXES): + return ProgrammingError(_ASYNC_DRIVER_ERROR, message) + return error + + +def _classify_database_error(error: Exception, default_type: type[DatabaseError]): + diagnostics = getattr(error, "sql_errors", ()) + numbers = {item.get("number") for item in diagnostics if isinstance(item, dict)} + if numbers & _DATA_ERROR_NUMBERS: + return DataError + if numbers & _INTEGRITY_ERROR_NUMBERS: + return IntegrityError + if numbers & _PROGRAMMING_ERROR_NUMBERS: + return ProgrammingError + return default_type + def translate_py_core_exception(error: Exception) -> Exception: - """Return the equivalent public exception, or the original non-py-core error.""" + """Translate py-core and recognized built-in errors to public exceptions.""" for error_type in type(error).__mro__: if error_type.__module__ != "mssql_py_core": continue public_type = _EXCEPTION_TYPES.get(error_type.__name__) if public_type is None: continue + if public_type is DatabaseError: + public_type = _classify_database_error(error, public_type) logger.debug( "Async exception translation: %s -> %s", @@ -53,12 +99,12 @@ def translate_py_core_exception(error: Exception) -> Exception: setattr(translated, attribute, getattr(error, attribute)) return translated - return error + return _translate_known_builtin_error(error) @contextmanager def translate_py_core_exceptions() -> Iterator[None]: - """Translate only exceptions originating from mssql-py-core.""" + """Translate py-core and recognized built-in errors raised by the wrapped operation.""" try: yield except Exception as error: diff --git a/tests/AsyncTest/test_002_async_connection.py b/tests/AsyncTest/test_002_async_connection.py index 1067d46e5..ef0b10022 100644 --- a/tests/AsyncTest/test_002_async_connection.py +++ b/tests/AsyncTest/test_002_async_connection.py @@ -2,7 +2,11 @@ pytest.importorskip("mssql_py_core", exc_type=ImportError) -from mssql_python import ConnectionStringParseError, InterfaceError, NotSupportedError +from mssql_python import ( + ConnectionStringParseError, + InterfaceError, + NotSupportedError, +) from mssql_python.async_query import AsyncConnection, AsyncCursor from mssql_python.async_query._connection_context import build_async_connection_context from mssql_python.helpers import connstr_to_pycore_params @@ -309,18 +313,25 @@ async def test_close_can_be_called_repeatedly(async_connection_string): @pytest.mark.asyncio -async def test_operations_after_close_preserve_native_errors(async_connection_string): +async def test_operations_after_close_translate_native_errors(async_connection_string): connection = await AsyncConnection.connect(async_connection_string) await connection.close() - with pytest.raises(RuntimeError, match="Connection is closed"): + with pytest.raises(InterfaceError, match="Connection is closed") as cursor_error: connection.cursor() - with pytest.raises(RuntimeError, match="Connection is closed"): + assert isinstance(cursor_error.value.__cause__, RuntimeError) + + with pytest.raises(InterfaceError, match="Connection is closed") as commit_error: await connection.commit() - with pytest.raises(RuntimeError, match="Connection is closed"): + assert isinstance(commit_error.value.__cause__, RuntimeError) + + with pytest.raises(InterfaceError, match="Connection is closed") as rollback_error: await connection.rollback() - with pytest.raises(RuntimeError, match="Connection is closed"): + assert isinstance(rollback_error.value.__cause__, RuntimeError) + + with pytest.raises(InterfaceError, match="Connection is closed") as enter_error: await connection.__aenter__() + assert isinstance(enter_error.value.__cause__, RuntimeError) @pytest.mark.asyncio diff --git a/tests/AsyncTest/test_003_async_exceptions.py b/tests/AsyncTest/test_003_async_exceptions.py index c4f16aec3..37081f7d2 100644 --- a/tests/AsyncTest/test_003_async_exceptions.py +++ b/tests/AsyncTest/test_003_async_exceptions.py @@ -1,4 +1,5 @@ from typing import Any, cast +from uuid import uuid4 import pytest @@ -59,12 +60,54 @@ def test_translation_preserves_native_diagnostic_attributes(): assert translated.info_messages == native_error.info_messages +@pytest.mark.parametrize( + "number, public_type", + ( + (8134, public_exceptions.DataError), + (2627, public_exceptions.IntegrityError), + (156, public_exceptions.ProgrammingError), + (50001, public_exceptions.DatabaseError), + ), +) +def test_translation_classifies_sql_server_diagnostics(number, public_type): + native_error = getattr(mssql_py_core, "DatabaseError")("query failed") + native_error.sql_errors = [{"number": number}] + + translated = translate_py_core_exception(native_error) + + assert type(translated) is public_type + assert cast(Any, translated).sql_errors == native_error.sql_errors + + def test_non_py_core_exception_is_not_translated(): error = RuntimeError("unrelated failure") assert translate_py_core_exception(error) is error +@pytest.mark.parametrize( + "native_error, public_type", + ( + (RuntimeError("Cursor is closed"), public_exceptions.ProgrammingError), + (RuntimeError("Connection is closed"), public_exceptions.InterfaceError), + (RuntimeError("Connection is closing"), public_exceptions.InterfaceError), + ( + RuntimeError("Connection is busy with another cursor operation"), + public_exceptions.OperationalError, + ), + (RuntimeError("Connection is broken"), public_exceptions.OperationalError), + ( + TypeError("The SQL contains 2 parameter markers, but 1 parameters were supplied"), + public_exceptions.ProgrammingError, + ), + ), +) +def test_translates_known_py_core_builtin_errors(native_error, public_type): + translated = translate_py_core_exception(native_error) + + assert isinstance(translated, public_type) + + def test_translation_context_preserves_native_error_as_cause(): native_error = getattr(mssql_py_core, "OperationalError")("connection lost") @@ -76,13 +119,22 @@ def test_translation_context_preserves_native_error_as_cause(): @pytest.mark.asyncio -async def test_sql_error_translation_preserves_server_diagnostics(async_connection): +@pytest.mark.parametrize( + "fetch", + ( + lambda cursor: cursor.fetchone(), + lambda cursor: cursor.fetchall(), + lambda cursor: cursor.fetchmany(), + ), + ids=("fetchone", "fetchall", "fetchmany"), +) +async def test_fetch_data_error_preserves_server_diagnostics(async_connection, fetch): cursor = async_connection.cursor() try: await cursor.execute("SELECT 1 / 0") - with pytest.raises(public_exceptions.DatabaseError) as caught: - await cursor.fetchone() + with pytest.raises(public_exceptions.DataError) as caught: + await fetch(cursor) assert type(caught.value.__cause__) is getattr(mssql_py_core, "DatabaseError") assert getattr(caught.value, "sql_errors")[0]["number"] == 8134 @@ -90,6 +142,176 @@ async def test_sql_error_translation_preserves_server_diagnostics(async_connecti await cursor.close() +@pytest.mark.asyncio +@pytest.mark.parametrize( + "operation", + ( + lambda cursor: cursor.execute("SELECT 1"), + lambda cursor: cursor.executemany("SELECT ?", [(1,)]), + lambda cursor: cursor.fetchone(), + lambda cursor: cursor.fetchall(), + lambda cursor: cursor.fetchmany(), + lambda cursor: cursor.fetchmany(0), + lambda cursor: cursor.fetchmany("invalid"), + ), + ids=( + "execute", + "executemany", + "fetchone", + "fetchall", + "fetchmany", + "fetchmany-zero", + "fetchmany-invalid", + ), +) +async def test_closed_cursor_operations_raise_programming_error(async_connection, operation): + cursor = async_connection.cursor() + await cursor.close() + + with pytest.raises(public_exceptions.ProgrammingError) as caught: + await operation(cursor) + + assert isinstance(caught.value.__cause__, RuntimeError) + + +@pytest.mark.asyncio +async def test_closed_cursor_executemany_checks_state_before_parameters(async_connection): + cursor = async_connection.cursor() + await cursor.close() + + with pytest.raises(public_exceptions.ProgrammingError) as caught: + await cursor.executemany("SELECT ?", iter([(1,)])) + + assert isinstance(caught.value.__cause__, RuntimeError) + + +@pytest.mark.asyncio +async def test_connection_close_invalidates_cursor_fetchmany_fast_path(async_connection_string): + connection = await AsyncConnection.connect(async_connection_string) + cursor = connection.cursor() + await connection.close() + + with pytest.raises(public_exceptions.InterfaceError) as caught: + await cursor.fetchmany(0) + + assert isinstance(caught.value.__cause__, RuntimeError) + + +@pytest.mark.asyncio +async def test_execute_parameter_count_error_is_programming_error_and_cursor_is_reusable( + async_cursor, +): + with pytest.raises(public_exceptions.ProgrammingError) as caught: + await async_cursor.execute("SELECT ?, ?", 1) + + assert isinstance(caught.value.__cause__, TypeError) + assert await async_cursor.execute("SELECT 1") is async_cursor + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "fetch", + ( + lambda cursor: cursor.fetchone(), + lambda cursor: cursor.fetchall(), + lambda cursor: cursor.fetchmany(), + ), + ids=("fetchone", "fetchall", "fetchmany"), +) +async def test_fetch_without_result_set_raises_programming_error(async_cursor, fetch): + with pytest.raises(public_exceptions.ProgrammingError) as caught: + await fetch(async_cursor) + + assert type(caught.value.__cause__) is getattr(mssql_py_core, "ProgrammingError") + assert str(caught.value.__cause__) == "No active result set" + + +@pytest.mark.asyncio +async def test_client_validation_errors_remain_native_python_errors(async_cursor): + with pytest.raises(KeyError): + await async_cursor.execute("SELECT %(missing)s", {"other": 1}) + + with pytest.raises(TypeError): + await async_cursor.executemany("SELECT %(value)s", [{"value": 1}, (2,)]) + + await async_cursor.execute("SELECT 1") + with pytest.raises(TypeError): + await async_cursor.fetchmany("invalid") + + +@pytest.mark.asyncio +async def test_execute_programming_error_preserves_diagnostics_and_cursor_is_reusable( + async_cursor, +): + with pytest.raises(public_exceptions.ProgrammingError) as caught: + await async_cursor.execute("SELECT FROM") + + assert getattr(caught.value, "sql_errors")[0]["number"] == 156 + assert await async_cursor.execute("SELECT 1") is async_cursor + + +@pytest.mark.asyncio +async def test_executemany_integrity_error_reports_row_and_preserves_partial_progress( + async_connection_string, +): + connection = await AsyncConnection.connect(async_connection_string, autocommit=True) + cursor = connection.cursor() + table_name = f"async_exception_{uuid4().hex}" + try: + await cursor.execute(f"CREATE TABLE {table_name} (id INT PRIMARY KEY)") + + with pytest.raises(public_exceptions.IntegrityError) as caught: + await cursor.executemany( + f"INSERT INTO {table_name} VALUES (?)", + [(1,), (1,), (2,)], + ) + + assert "parameter row 1" in str(caught.value) + assert getattr(caught.value, "sql_errors")[0]["number"] == 2627 + await cursor.execute(f"SELECT COUNT(*) FROM {table_name}") + row = await cursor.fetchone() + assert row is not None + assert row[0] == 1 + finally: + await cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + await cursor.close() + await connection.close() + + +@pytest.mark.asyncio +async def test_timeout_is_operational_error_and_cursor_is_reusable(async_connection_string): + connection = await AsyncConnection.connect(async_connection_string) + connection.timeout = 1 + cursor = connection.cursor() + try: + with pytest.raises(public_exceptions.OperationalError) as caught: + await cursor.execute("WAITFOR DELAY '00:00:03'; SELECT 1") + + assert type(caught.value.__cause__) is getattr(mssql_py_core, "OperationalError") + assert await cursor.execute("SELECT 1") is cursor + finally: + await cursor.close() + await connection.close() + + +@pytest.mark.asyncio +async def test_busy_connection_is_operational_error(async_connection): + owning_cursor = async_connection.cursor() + blocked_cursor = async_connection.cursor() + try: + await owning_cursor.execute("SELECT 1 UNION ALL SELECT 2") + + with pytest.raises(public_exceptions.OperationalError) as caught: + await blocked_cursor.execute("SELECT 3") + + assert isinstance(caught.value.__cause__, RuntimeError) + await owning_cursor.close() + assert await blocked_cursor.execute("SELECT 3") is blocked_cursor + finally: + await owning_cursor.close() + await blocked_cursor.close() + + def test_async_connection_exposes_public_exception_classes(): for name in EXCEPTION_NAMES: assert getattr(AsyncConnection, name) is getattr(public_exceptions, name) diff --git a/tests/AsyncTest/test_004_async_logging.py b/tests/AsyncTest/test_004_async_logging.py index 23df85147..2e16f0eb6 100644 --- a/tests/AsyncTest/test_004_async_logging.py +++ b/tests/AsyncTest/test_004_async_logging.py @@ -7,6 +7,9 @@ from mssql_python import OperationalError, setup_logging from mssql_python.async_query import AsyncConnection from mssql_python.async_query import exception_translator +from mssql_python.connection_string_parser import ( + _ConnectionStringParser, # pyright: ignore[reportPrivateUsage] +) from mssql_python.logging import logger @@ -36,6 +39,14 @@ def enable_file_logging(tmp_path, name): return log_path +def log_contains_connection_password(messages, connection_string): + connection_params = _ConnectionStringParser()._parse( # pyright: ignore[reportPrivateUsage] + str(connection_string) + ) + password = connection_params.get("pwd") or connection_params.get("password") + return bool(password and password in messages) + + @pytest.mark.asyncio async def test_connect_logging_does_not_include_client_context( async_connection_string, @@ -50,8 +61,46 @@ async def test_connect_logging_does_not_include_client_context( assert "AsyncConnection.connect: starting" in messages assert "AsyncConnection.connect: connected" in messages assert "PWD=" not in messages - assert "password" not in messages.lower() + assert "password=" not in messages.lower() assert "client_context" not in messages + if log_contains_connection_password(messages, async_connection_string): + pytest.fail("Async connection logs contain the SQL authentication secret") + + +@pytest.mark.asyncio +async def test_default_logger_combines_python_and_py_core_operation_logs( + async_connection_string, + tmp_path, +): + log_path = enable_file_logging(tmp_path, "async-operations.log") + connection = await AsyncConnection.connect(async_connection_string, autocommit=True) + cursor = connection.cursor() + try: + await cursor.execute("SELECT CAST(? AS INT) AS value UNION ALL SELECT 2", 1) + await cursor.fetchone() + await cursor.fetchmany(1) + await cursor.execute("SELECT CAST(3 AS INT) AS value") + await cursor.fetchall() + await cursor.executemany("SELECT CAST(? AS INT)", [(4,), (5,)]) + finally: + await cursor.close() + await connection.close() + + messages = read_log(log_path) + expected_python_messages = ( + "AsyncCursor.execute: starting; param_count=1", + "AsyncCursor.execute: completed; rowcount=-1; column_count=1; has_result_set=True", + "AsyncCursor.fetchone: completed; row_found=True; rowcount=1", + "AsyncCursor.fetchmany: starting; requested_size=1", + "AsyncCursor.fetchmany: completed; row_count=1; rowcount=2", + "AsyncCursor.fetchall: completed; row_count=1; rowcount=1", + "AsyncCursor.executemany: starting; batch_count=2", + "AsyncCursor.executemany: completed; rowcount=-1", + ) + for expected in expected_python_messages: + assert expected in messages + assert ", py-core, " in messages + assert "PWD=" not in messages @pytest.mark.asyncio diff --git a/tests/AsyncTest/test_005_async_cursor.py b/tests/AsyncTest/test_005_async_cursor.py index 6a7c4fec1..80eabc962 100644 --- a/tests/AsyncTest/test_005_async_cursor.py +++ b/tests/AsyncTest/test_005_async_cursor.py @@ -1,111 +1,10 @@ -from typing import Any, cast -from uuid import uuid4 - import pytest pytest.importorskip("mssql_py_core", exc_type=ImportError) -from mssql_python import DatabaseError -from mssql_python.async_query import AsyncCursor - - -@pytest.mark.asyncio -@pytest.mark.parametrize("use_prepare", (True, False)) -async def test_execute_returns_public_cursor_and_binds_parameters( - async_connection, - use_prepare, -): - cursor = async_connection.cursor() - try: - result = await cursor.execute( - "SELECT CAST(? AS INT) AS value", - 7, - use_prepare=use_prepare, - reset_cursor=False, - ) - - assert result is cursor - assert await cursor.fetchone() == (7,) - finally: - await cursor.close() - - -@pytest.mark.asyncio -@pytest.mark.parametrize("parameters", ((1, 2), [1, 2])) -@pytest.mark.parametrize("use_prepare", (True, False)) -async def test_execute_accepts_single_parameter_sequence( - async_cursor, - parameters, - use_prepare, -): - await async_cursor.execute( - "SELECT CAST(? AS INT), CAST(? AS INT)", - parameters, - use_prepare=use_prepare, - ) - - assert await async_cursor.fetchone() == (1, 2) - - -@pytest.mark.asyncio -async def test_executemany_returns_public_cursor_and_inserts_rows(async_connection): - cursor = async_connection.cursor() - rows = [(1, "one"), (2, "two")] - table_name = f"async_cursor_test_{uuid4().hex}" - try: - await cursor.execute( - f"CREATE TABLE {table_name} (id INT NOT NULL, value NVARCHAR(20) NOT NULL)" - ) - result = await cursor.executemany( - f"INSERT INTO {table_name} (id, value) VALUES (?, ?)", - rows, - use_prepare=False, - ) - assert result is cursor - - await cursor.execute(f"SELECT id, value FROM {table_name} ORDER BY id") - assert await cursor.fetchall() == rows - finally: - await cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - await cursor.close() - @pytest.mark.asyncio -async def test_fetch_and_result_navigation_preserve_native_values(async_connection): - cursor = async_connection.cursor() - try: - await cursor.execute( - "SELECT CAST(1 AS INT) AS value UNION ALL SELECT 2 ORDER BY value; " - "SELECT CAST(3 AS INT) AS value" - ) - - assert await cursor.fetchone() == (1,) - assert await cursor.fetchmany(1) == [(2,)] - assert await cursor.fetchall() == [] - assert await cursor.nextset() is True - assert await cursor.fetchone() == (3,) - assert await cursor.nextset() is False - finally: - await cursor.close() - - -@pytest.mark.asyncio -async def test_fetchmany_uses_arraysize(async_connection): - cursor = async_connection.cursor() - try: - cursor.arraysize = 2 - await cursor.execute( - "SELECT CAST(1 AS INT) AS value UNION ALL SELECT 2 UNION ALL SELECT 3 ORDER BY value" - ) - - assert await cursor.fetchmany() == [(1,), (2,)] - assert await cursor.fetchall() == [(3,)] - finally: - await cursor.close() - - -@pytest.mark.asyncio -async def test_properties_and_setinputsizes_use_native_cursor(async_connection): +async def test_properties_and_setinputsizes_use_py_core_async_cursor(async_connection): cursor = async_connection.cursor() try: assert cursor.timeout == async_connection.timeout @@ -115,12 +14,13 @@ async def test_properties_and_setinputsizes_use_native_cursor(async_connection): cursor.arraysize = 50 cursor.setinputsizes([(4, 10, 0)]) - await cursor.execute("SELECT CAST(? AS INT) AS value", 9) + await cursor.execute( + "IF CAST(? AS INT) <> 9 THROW 50000, 'Unexpected parameter value', 1", + 9, + ) assert cursor.arraysize == 50 - description = cast(Any, cursor.description) - assert description[0][0] == "value" - assert await cursor.fetchone() == (9,) + assert cursor.description is None finally: await cursor.close() @@ -134,18 +34,12 @@ async def test_close_is_idempotent(async_connection): @pytest.mark.asyncio -async def test_cursor_operation_translates_native_exception(async_connection): +async def test_close_clears_cached_fetch_rowcount(async_connection): cursor = async_connection.cursor() - try: - await cursor.execute("SELECT 1 / 0") + await cursor.execute("SELECT 1 AS value") + await cursor.fetchone() + assert cursor.rowcount == 1 - with pytest.raises(DatabaseError) as caught: - await cursor.fetchone() + await cursor.close() - assert type(caught.value.__cause__).__module__ == "mssql_py_core" - assert type(caught.value.__cause__).__name__ == "DatabaseError" - sql_errors = getattr(caught.value, "sql_errors") - assert sql_errors - assert sql_errors[0]["number"] == 8134 - finally: - await cursor.close() + assert cursor.rowcount == -1 diff --git a/tests/AsyncTest/test_006_async_execute.py b/tests/AsyncTest/test_006_async_execute.py new file mode 100644 index 000000000..3fbdd0260 --- /dev/null +++ b/tests/AsyncTest/test_006_async_execute.py @@ -0,0 +1,304 @@ +from datetime import date, datetime, time +from decimal import Decimal +import pytest +from uuid import UUID, uuid4 + +from mssql_python.constants import ConstantsDDBC + +pytest.importorskip("mssql_py_core", exc_type=ImportError) + +from mssql_python.async_query import AsyncConnection, AsyncCursor, async_execute +from mssql_python.row import Row + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (True, False)) +async def test_execute_returns_public_cursor_and_binds_parameters( + async_connection, + use_prepare, +): + assert isinstance(async_connection, AsyncConnection) + + cursor = async_connection.cursor() + try: + assert isinstance(cursor, AsyncCursor) + + result = await cursor.execute( + "IF CAST(? AS INT) <> 7 THROW 50000, 'Unexpected parameter value', 1", + 7, + use_prepare=use_prepare, + reset_cursor=False, + ) + + assert result is cursor + finally: + await cursor.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("parameters", ((1, 2), [1, 2])) +@pytest.mark.parametrize("use_prepare", (True, False)) +async def test_execute_accepts_single_parameter_sequence( + async_cursor, + parameters, + use_prepare, +): + await async_cursor.execute( + "IF CAST(? AS INT) <> 1 OR CAST(? AS INT) <> 2 " + "THROW 50000, 'Unexpected parameter values', 1", + parameters, + use_prepare=use_prepare, + ) + + +@pytest.mark.asyncio +async def test_execute_accepts_named_parameters(async_cursor): + result = await async_cursor.execute( + "IF CAST(%(first)s AS INT) <> 1 OR CAST(%(second)s AS INT) <> 2 " + "THROW 50000, 'Unexpected parameter values', 1", + {"first": 1, "second": 2}, + ) + + assert result is async_cursor + + +@pytest.mark.asyncio +async def test_execute_accepts_dbapi_row(async_cursor, monkeypatch): + row = Row([1, 2], {"first": 0, "second": 1}) + log_calls = [] + monkeypatch.setattr( + async_execute.logger, + "debug", + lambda message, *args: log_calls.append((message, args)), + ) + + result = await async_cursor.execute( + "IF CAST(? AS INT) <> 1 OR CAST(? AS INT) <> 2 " + "THROW 50000, 'Unexpected parameter values', 1", + row, + ) + + assert result is async_cursor + assert log_calls[0][1][0] == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (True, False)) +@pytest.mark.parametrize( + "operation, rows", + ( + ("INSERT INTO {table} (id, value) VALUES (?, ?)", [(1, "one"), (2, "two")]), + ( + "INSERT INTO {table} (id, value) VALUES (%(id)s, %(value)s)", + [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}], + ), + ), +) +async def test_executemany_matches_sync_contract( + async_connection, + operation, + rows, + use_prepare, +): + cursor = async_connection.cursor() + table_name = f"async_execute_test_{uuid4().hex}" + try: + await cursor.execute( + f"CREATE TABLE {table_name} (id INT NOT NULL, value NVARCHAR(20) NOT NULL)" + ) + result = await cursor.executemany( + operation.format(table=table_name), + rows, + use_prepare=use_prepare, + ) + assert result is None + assert cursor.rowcount == 2 + finally: + await cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + await cursor.close() + + +@pytest.mark.asyncio +async def test_executemany_rejects_non_sequence_like_sync(async_cursor): + with pytest.raises(TypeError): + await async_cursor.executemany("SELECT CAST(? AS INT)", iter([(1,), (2,)])) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "value, sql_type", + ( + (True, "BIT"), + (-(2**63), "BIGINT"), + (2**63 - 1, "BIGINT"), + (3.25, "FLOAT"), + ("hello\x00world", "NVARCHAR(20)"), + (b"\x00\x01\xff", "VARBINARY(20)"), + (Decimal("123.45"), "DECIMAL(10, 2)"), + (date(2024, 1, 2), "DATE"), + (time(12, 34, 56, 123456), "TIME(6)"), + (datetime(2024, 1, 2, 12, 34, 56, 123456), "DATETIME2(6)"), + (UUID("6f9619ff-8b86-d011-b42d-00c04fc964ff"), "UNIQUEIDENTIFIER"), + ), +) +async def test_execute_binds_representative_sync_parameter_types(async_cursor, value, sql_type): + result = await async_cursor.execute( + f"IF CAST(? AS {sql_type}) IS NULL THROW 50000, 'Unexpected NULL value', 1", + value, + ) + + assert result is async_cursor + + +@pytest.mark.asyncio +async def test_execute_does_not_mutate_caller_parameter_list(async_cursor): + parameters = ["hello", 42, Decimal("3.14"), date(2024, 1, 1)] + snapshot = list(parameters) + + await async_cursor.execute( + "IF CAST(? AS NVARCHAR(10)) <> 'hello' OR CAST(? AS INT) <> 42 " + "OR CAST(? AS DECIMAL(4, 2)) <> 3.14 OR CAST(? AS DATE) <> '2024-01-01' " + "THROW 50000, 'Unexpected parameter values', 1", + parameters, + ) + + assert parameters == snapshot + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", ({1, 2, 3}, Decimal("NaN"), Decimal("Infinity"))) +async def test_execute_rejects_unsupported_or_non_finite_parameters(async_cursor, value): + with pytest.raises((TypeError, ValueError)): + await async_cursor.execute("SELECT ?", value) + + +@pytest.mark.asyncio +async def test_execute_reset_cursor_false_supports_repeated_execution(async_cursor): + for value in range(5): + result = await async_cursor.execute( + "IF CAST(? AS INT) < 0 THROW 50000, 'Unexpected parameter value', 1", + value, + reset_cursor=value == 0, + ) + + assert result is async_cursor + + +@pytest.mark.asyncio +async def test_execute_updates_rowcount_and_description(async_cursor): + table_name = f"async_execute_state_{uuid4().hex}" + try: + await async_cursor.execute(f"CREATE TABLE {table_name} (value INT)") + await async_cursor.execute(f"INSERT INTO {table_name} VALUES (1), (2), (3)") + assert async_cursor.rowcount == 3 + assert async_cursor.description is None + + await async_cursor.execute(f"SELECT value AS named_value FROM {table_name}") + assert async_cursor.rowcount == -1 + description = async_cursor.description + assert description is not None + assert description[0][0] == "named_value" + finally: + await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + + +@pytest.mark.asyncio +async def test_execute_repeated_null_parameters(async_cursor): + table_name = f"async_execute_nulls_{uuid4().hex}" + try: + await async_cursor.execute(f"CREATE TABLE {table_name} (id INT, value VARCHAR(20))") + for identifier in range(1, 4): + await async_cursor.execute( + f"INSERT INTO {table_name} VALUES (?, ?)", + identifier, + None, + reset_cursor=identifier == 1, + ) + + assert async_cursor.rowcount == 1 + await async_cursor.execute( + f"IF (SELECT COUNT(*) FROM {table_name} WHERE value IS NULL) <> 3 " + "THROW 50000, 'Unexpected NULL count', 1" + ) + finally: + await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + + +@pytest.mark.asyncio +async def test_executemany_empty_sequence_sets_rowcount_zero(async_cursor): + result = await async_cursor.executemany("SELECT CAST(? AS INT)", []) + + assert result is None + assert async_cursor.rowcount == 0 + + +@pytest.mark.asyncio +async def test_executemany_handles_sync_edge_value_batches(async_cursor): + table_name = f"async_many_values_{uuid4().hex}" + try: + await async_cursor.execute( + f"CREATE TABLE {table_name} (" + "id INT, text_value NVARCHAR(50), binary_value VARBINARY(20), " + "integer_value BIGINT, decimal_value DECIMAL(18, 10), date_value DATE)" + ) + rows = [ + (1, "", b"", -(2**63), Decimal("-1.25"), date(2024, 1, 1)), + (2, None, None, 0, None, None), + ( + 3, + "unicode-\u03bb", + b"\x00\xff", + 2**63 - 1, + Decimal("999.99"), + date(2024, 1, 3), + ), + ] + async_cursor.setinputsizes( + [ + ConstantsDDBC.SQL_INTEGER.value, + (ConstantsDDBC.SQL_WVARCHAR.value, 50, 0), + (ConstantsDDBC.SQL_VARBINARY.value, 20, 0), + ConstantsDDBC.SQL_BIGINT.value, + (ConstantsDDBC.SQL_DECIMAL.value, 18, 10), + ConstantsDDBC.SQL_TYPE_DATE.value, + ] + ) + + result = await async_cursor.executemany( + f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?, ?)", + rows, + ) + + assert result is None + assert async_cursor.rowcount == len(rows) + await async_cursor.execute( + f"IF (SELECT COUNT(*) FROM {table_name}) <> 3 " + f"OR (SELECT COUNT(*) FROM {table_name} WHERE text_value = '') <> 1 " + f"OR (SELECT COUNT(*) FROM {table_name} WHERE text_value IS NULL) <> 1 " + f"OR (SELECT COUNT(*) FROM {table_name} WHERE binary_value = 0x) <> 1 " + "THROW 50000, 'Unexpected batch values', 1" + ) + finally: + async_cursor.setinputsizes(None) + await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + + +@pytest.mark.asyncio +async def test_executemany_handles_multiple_all_null_columns(async_cursor): + table_name = f"async_many_nulls_{uuid4().hex}" + try: + await async_cursor.execute( + f"CREATE TABLE {table_name} (id INT, text_value VARCHAR(20), number_value INT)" + ) + rows = [(1, None, None), (2, None, None), (3, None, None)] + + await async_cursor.executemany(f"INSERT INTO {table_name} VALUES (?, ?, ?)", rows) + + assert async_cursor.rowcount == len(rows) + await async_cursor.execute( + f"IF (SELECT COUNT(*) FROM {table_name} " + "WHERE text_value IS NULL AND number_value IS NULL) <> 3 " + "THROW 50000, 'Unexpected NULL values', 1" + ) + finally: + await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") diff --git a/tests/AsyncTest/test_007_async_fetch.py b/tests/AsyncTest/test_007_async_fetch.py new file mode 100644 index 000000000..910bc8a36 --- /dev/null +++ b/tests/AsyncTest/test_007_async_fetch.py @@ -0,0 +1,341 @@ +import pytest +from datetime import date, datetime, time +from decimal import Decimal +from uuid import UUID + +pytest.importorskip("mssql_py_core", exc_type=ImportError) + +import mssql_python +from mssql_python import DataError, Row +from mssql_python.async_query import AsyncCursor + + +@pytest.mark.asyncio +async def test_fetch_and_result_navigation_preserve_native_values(async_connection): + cursor = async_connection.cursor() + try: + await cursor.execute( + "SELECT CAST(1 AS INT) AS value UNION ALL SELECT 2 ORDER BY value; " + "SELECT CAST(3 AS INT) AS value" + ) + + first = await cursor.fetchone() + assert isinstance(first, Row) + assert tuple(first) == (1,) + assert first.value == 1 + assert cursor.rowcount == 1 + + many = await cursor.fetchmany(1) + assert all(isinstance(row, Row) for row in many) + assert [tuple(row) for row in many] == [(2,)] + assert cursor.rowcount == 2 + assert await cursor.fetchall() == [] + assert await cursor.fetchone() is None + assert cursor.rowcount == 2 + assert await cursor.nextset() is True + next_result = await cursor.fetchone() + assert isinstance(next_result, Row) + assert tuple(next_result) == (3,) + assert await cursor.nextset() is False + assert cursor.description is None + finally: + await cursor.close() + + +@pytest.mark.asyncio +async def test_nextset_failure_clears_previous_result_state(): + class FailingNativeCursor: + rowcount = -1 + + async def nextset(self): + raise RuntimeError("nextset failed") + + class StatefulAsyncCursor(AsyncCursor): + def seed_result_state(self): + self._description = [("value", int, None, None, None, None, True)] + self._column_map = {"value": 0} + self._column_map_lower = {"value": 0} + self._uuid_str_indices = (0,) + self._fetched_row_count = 2 + self._fetch_rowcount = 2 + + def result_maps(self): + return self._column_map, self._column_map_lower, self._uuid_str_indices + + cursor = StatefulAsyncCursor(FailingNativeCursor()) + cursor.seed_result_state() + + with pytest.raises(RuntimeError, match="nextset failed"): + await cursor.nextset() + + assert cursor.description is None + assert cursor.rowcount == -1 + assert cursor.result_maps() == ({}, None, None) + + +@pytest.mark.asyncio +async def test_fetchmany_uses_arraysize(async_connection): + cursor = async_connection.cursor() + try: + cursor.arraysize = 2 + await cursor.execute( + "SELECT CAST(1 AS INT) AS value UNION ALL SELECT 2 UNION ALL SELECT 3 ORDER BY value" + ) + + assert await cursor.fetchmany(0) == [] + assert await cursor.fetchmany(-1) == [] + assert [tuple(row) for row in await cursor.fetchmany()] == [(1,), (2,)] + assert [tuple(row) for row in await cursor.fetchall()] == [(3,)] + finally: + await cursor.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("size", (0, -1)) +async def test_fetchmany_non_positive_size_skips_native_fetch(async_connection, monkeypatch, size): + cursor = async_connection.cursor() + try: + await cursor.execute("SELECT 1 AS value") + + class UnexpectedNativeFetch: + async def fetchmany(self, *_args): + pytest.fail("fetchmany must not call py-core for a non-positive size") + + monkeypatch.setattr( + "mssql_python.async_query.async_fetch._get_py_core_async_cursor", + lambda _cursor: UnexpectedNativeFetch(), + ) + + assert await cursor.fetchmany(size) == [] + assert cursor.rowcount == -1 + finally: + await cursor.close() + + +@pytest.mark.asyncio +async def test_fetch_respects_row_settings(async_connection): + cursor = async_connection.cursor() + previous_lowercase = mssql_python.lowercase + previous_native_uuid = mssql_python.native_uuid + try: + mssql_python.lowercase = True + mssql_python.native_uuid = False + await cursor.execute( + "SELECT CAST('6F9619FF-8B86-D011-B42D-00C04FC964FF' " + "AS UNIQUEIDENTIFIER) AS MixedGuid" + ) + + row = await cursor.fetchone() + + assert isinstance(row, Row) + assert cursor.description[0][0] == "mixedguid" + assert row.MixedGuid == "6F9619FF-8B86-D011-B42D-00C04FC964FF" + assert row.mixedguid == row.MixedGuid + assert row.MIXEDGUID == row.MixedGuid + finally: + mssql_python.lowercase = previous_lowercase + mssql_python.native_uuid = previous_native_uuid + await cursor.close() + + +@pytest.mark.asyncio +async def test_row_settings_are_snapshotted_at_execute(async_connection): + cursor = async_connection.cursor() + previous_lowercase = mssql_python.lowercase + previous_native_uuid = mssql_python.native_uuid + try: + mssql_python.lowercase = True + mssql_python.native_uuid = False + await cursor.execute( + "SELECT CAST('6F9619FF-8B86-D011-B42D-00C04FC964FF' " + "AS UNIQUEIDENTIFIER) AS MixedGuid UNION ALL " + "SELECT CAST('6F9619FF-8B86-D011-B42D-00C04FC964FE' AS UNIQUEIDENTIFIER)" + ) + + mssql_python.lowercase = False + mssql_python.native_uuid = True + first = await cursor.fetchone() + mssql_python.lowercase = True + mssql_python.native_uuid = False + second = await cursor.fetchone() + + assert cursor.description[0][0] == "mixedguid" + assert isinstance(first[0], str) + assert isinstance(second[0], str) + assert first.MixedGuid == first.mixedguid + assert second.MixedGuid == second.mixedguid + finally: + mssql_python.lowercase = previous_lowercase + mssql_python.native_uuid = previous_native_uuid + await cursor.close() + + +@pytest.mark.asyncio +async def test_row_settings_are_resnapshotted_at_nextset(async_connection): + cursor = async_connection.cursor() + previous_lowercase = mssql_python.lowercase + previous_native_uuid = mssql_python.native_uuid + try: + mssql_python.lowercase = False + mssql_python.native_uuid = True + await cursor.execute( + "SELECT CAST('6F9619FF-8B86-D011-B42D-00C04FC964FF' " + "AS UNIQUEIDENTIFIER) AS FirstGuid; " + "SELECT CAST('6F9619FF-8B86-D011-B42D-00C04FC964FE' " + "AS UNIQUEIDENTIFIER) AS SecondGuid" + ) + + mssql_python.lowercase = True + mssql_python.native_uuid = False + first = await cursor.fetchone() + assert cursor.description[0][0] == "FirstGuid" + assert isinstance(first[0], UUID) + + assert await cursor.nextset() is True + mssql_python.lowercase = False + mssql_python.native_uuid = True + second = await cursor.fetchone() + + assert cursor.description[0][0] == "secondguid" + assert isinstance(second[0], str) + assert second.SecondGuid == second.secondguid + finally: + mssql_python.lowercase = previous_lowercase + mssql_python.native_uuid = previous_native_uuid + await cursor.close() + + +@pytest.mark.asyncio +async def test_fetch_translates_py_core_exception(async_connection): + cursor = async_connection.cursor() + try: + await cursor.execute("SELECT 1 / 0") + + with pytest.raises(DataError) as caught: + await cursor.fetchone() + + assert type(caught.value.__cause__).__module__ == "mssql_py_core" + assert type(caught.value.__cause__).__name__ == "DatabaseError" + sql_errors = getattr(caught.value, "sql_errors") + assert sql_errors + assert sql_errors[0]["number"] == 8134 + finally: + await cursor.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "fetch_plan", + ( + ("many-one",), + ("one-many",), + ("many-one-many-one",), + ), +) +async def test_fetchone_fetchmany_interleaving(async_cursor, fetch_plan): + await async_cursor.execute( + "SELECT value FROM (VALUES (1), (2), (3), (4)) AS values_table(value) ORDER BY value" + ) + + if fetch_plan == ("many-one",): + assert [tuple(row) for row in await async_cursor.fetchmany(1)] == [(1,)] + assert tuple(await async_cursor.fetchone()) == (2,) + elif fetch_plan == ("one-many",): + assert tuple(await async_cursor.fetchone()) == (1,) + assert [tuple(row) for row in await async_cursor.fetchmany(2)] == [(2,), (3,)] + else: + assert [tuple(row) for row in await async_cursor.fetchmany(1)] == [(1,)] + assert tuple(await async_cursor.fetchone()) == (2,) + assert [tuple(row) for row in await async_cursor.fetchmany(1)] == [(3,)] + assert tuple(await async_cursor.fetchone()) == (4,) + + +@pytest.mark.asyncio +async def test_fetchmany_more_than_available_and_repeated_exhaustion(async_cursor): + await async_cursor.execute( + "SELECT value FROM (VALUES (1), (2), (3)) AS rows(value) ORDER BY value" + ) + + rows = await async_cursor.fetchmany(10) + + assert [row[0] for row in rows] == [1, 2, 3] + assert await async_cursor.fetchmany(10) == [] + assert await async_cursor.fetchmany(10) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fetch_method", ("fetchone", "fetchmany", "fetchall")) +async def test_fetch_empty_result_set(fetch_method, async_cursor): + await async_cursor.execute("SELECT CAST(1 AS INT) AS value WHERE 1 = 0") + + result = await getattr(async_cursor, fetch_method)() + + assert result is None if fetch_method == "fetchone" else result == [] + assert async_cursor.rowcount == 0 + + +@pytest.mark.asyncio +async def test_fetch_preserves_empty_string_binary_and_null(async_cursor): + await async_cursor.execute( + "SELECT * FROM (VALUES " + "(1, CAST('' AS NVARCHAR(10)), CAST(0x AS VARBINARY(10))), " + "(2, CAST(NULL AS NVARCHAR(10)), CAST(NULL AS VARBINARY(10))), " + "(3, CAST('text' AS NVARCHAR(10)), CAST(0x1234 AS VARBINARY(10)))) " + "AS values_table(id, text_value, binary_value) ORDER BY id" + ) + + first = await async_cursor.fetchone() + remaining = await async_cursor.fetchall() + + assert tuple(first) == (1, "", b"") + assert [tuple(row) for row in remaining] == [ + (2, None, None), + (3, "text", b"\x12\x34"), + ] + + +@pytest.mark.asyncio +async def test_fetchmany_handles_mixed_large_lob_sizes(async_cursor): + medium = "x" * 1_000 + large = "y" * 10_000 + await async_cursor.execute( + "SELECT * FROM (VALUES " + "(1, CAST('' AS NVARCHAR(MAX))), " + "(2, CAST(NULL AS NVARCHAR(MAX))), " + "(3, CAST(? AS NVARCHAR(MAX))), " + "(4, CAST(? AS NVARCHAR(MAX)))) AS values_table(id, value) ORDER BY id", + medium, + large, + ) + + first_batch = await async_cursor.fetchmany(3) + second_batch = await async_cursor.fetchmany(3) + + assert [row[1] for row in first_batch] == ["", None, medium] + assert [row[1] for row in second_batch] == [large] + assert await async_cursor.fetchmany(3) == [] + + +@pytest.mark.asyncio +async def test_fetch_roundtrips_representative_sync_result_types(async_cursor): + expected = ( + True, + -(2**63), + 3.25, + Decimal("123.45"), + date(2024, 1, 2), + time(12, 34, 56, 123456), + datetime(2024, 1, 2, 12, 34, 56, 123456), + UUID("6f9619ff-8b86-d011-b42d-00c04fc964ff"), + ) + await async_cursor.execute( + "SELECT CAST(1 AS BIT), CAST(-9223372036854775808 AS BIGINT), CAST(3.25 AS FLOAT), " + "CAST(123.45 AS DECIMAL(10, 2)), CAST('2024-01-02' AS DATE), " + "CAST('12:34:56.123456' AS TIME(6)), " + "CAST('2024-01-02T12:34:56.123456' AS DATETIME2(6)), " + "CAST('6F9619FF-8B86-D011-B42D-00C04FC964FF' AS UNIQUEIDENTIFIER)" + ) + + row = await async_cursor.fetchone() + + assert tuple(row) == expected