[Draft] Async Query execution using RUST - POC

.gitignore

@@ -84,3 +84,6 @@ fix_multiline_log_exclusions.py
 test_pyodbc_decimal.py
 run_coverage_docker.ps1
 TRIAGE_REPORT_*.md
+
+# Local-only async POC design docs (not committed upstream)
+docs/async/

mssql_python/__init__.py

@@ -60,6 +60,10 @@
 # Cursor Objects
 from .cursor import Cursor
 
+# Async POC surface (execute_async + fetchone_async only)
+from .connection_async import AsyncConnection, connect_async
+from .cursor_async import AsyncCursor
+
 # Row Objects
 from .row import Row
 

mssql_python/connection_async.py

@@ -0,0 +1,101 @@
+"""
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+
+Async connection for the mssql-py-core backend.
+
+Exposes :class:`AsyncConnection` and the awaitable factory
+:func:`connect_async`. Under the hood it drives
+``mssql_py_core.PyCoreAsyncConnection`` on a process-wide Tokio runtime.
+
+Design notes
+------------
+* Constructing :class:`AsyncConnection` currently connects *synchronously*
+  under the shared Tokio runtime. The method is exposed as an awaitable
+  factory (:meth:`AsyncConnection.connect`, :func:`connect_async`) so
+  callers can migrate to a truly-async connect path (planned) without
+  changing their code.
+
+* Concurrency contract — one physical TDS session, one in-flight batch.
+  Two concurrent ``await`` calls on the same :class:`AsyncConnection` (or
+  the same cursor) will *serialise* on the underlying
+  ``Arc<Mutex<TdsClient>>``. For true parallelism use multiple
+  :class:`AsyncConnection` instances.
+"""
+
+from __future__ import annotations
+
+import mssql_py_core
+
+from mssql_python.connection_string_parser import _ConnectionStringParser
+from mssql_python.cursor_async import AsyncCursor
+from mssql_python.exceptions import InterfaceError
+from mssql_python.helpers import connstr_to_pycore_params
+
+
+class AsyncConnection:
+    """Awaitable connection.
+
+    Constructing this object synchronously connects to the server on the
+    shared Tokio runtime. See :func:`connect_async` for a factory that
+    matches the awaitable idiom.
+    """
+
+    __slots__ = ("_core", "_closed")
+
+    def __init__(self, connection_str: str) -> None:
+        if not connection_str:
+            raise ValueError("connection_str must be a non-empty str")
+        parser = _ConnectionStringParser(validate_keywords=False)
+        params = parser._parse(connection_str)
+        pycore_ctx = connstr_to_pycore_params(params)
+        self._core = mssql_py_core.PyCoreAsyncConnection(pycore_ctx)
+        self._closed = False
+
+    @classmethod
+    async def connect(cls, connection_str: str) -> "AsyncConnection":
+        """Awaitable factory. Currently connects synchronously under the hood.
+
+        The method is ``async`` so callers can migrate to a truly-async
+        connect path (planned) without changing their code.
+        """
+        return cls(connection_str)
+
+    def cursor(self) -> AsyncCursor:
+        if self._closed:
+            raise InterfaceError(
+                driver_error="Connection is closed",
+                ddbc_error="AsyncConnection.cursor called after close",
+            )
+        return AsyncCursor(self._core.cursor())
+
+    async def close(self) -> None:
+        if not self._closed:
+            self._core.close()
+            self._closed = True
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    async def __aenter__(self) -> "AsyncConnection":
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.close()
+
+
+async def connect_async(connection_str: str) -> AsyncConnection:
+    """Awaitable factory for an :class:`AsyncConnection`.
+
+    Example::
+
+        async with await connect_async(conn_str) as conn:
+            async with conn.cursor() as cur:
+                await cur.execute("SELECT 1")
+                print(await cur.fetchone())
+    """
+    return await AsyncConnection.connect(connection_str)
+
+
+__all__ = ["AsyncConnection", "connect_async"]

mssql_python/cursor_async.py

@@ -0,0 +1,91 @@
+"""
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+
+Async cursor for the mssql-py-core backend.
+
+Wraps the Rust ``mssql_py_core.PyCoreAsyncCursor`` type in a small Python
+class so callers get proper coroutine methods (``async def foo(...)``) that
+work with ``asyncio.create_task`` / ``gather`` / ``wait_for``.
+
+The Rust ``execute_async`` / ``fetchone_async`` methods return a PyO3
+awaitable, not a coroutine — hence the ``async def foo(...): return await
+self._core.foo_async(...)`` pattern used here.
+
+POC scope: only ``execute`` and ``fetchone`` are async. Parameters are not
+yet supported.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from mssql_python.exceptions import InterfaceError
+
+
+class AsyncCursor:
+    """Awaitable cursor.
+
+    Not thread-safe — bind one to each asyncio task. Instances are created
+    via :meth:`mssql_python.connection_async.AsyncConnection.cursor`;
+    construct directly only in tests.
+    """
+
+    __slots__ = ("_core", "_closed")
+
+    def __init__(self, core_cursor: Any) -> None:
+        self._core = core_cursor
+        self._closed = False
+
+    async def execute(
+        self,
+        operation: str,
+        timeout_sec: int = 30,
+    ) -> "AsyncCursor":
+        """Execute a T-SQL batch. POC: parameters are not supported."""
+        if self._closed:
+            raise InterfaceError(
+                driver_error="Cursor is closed",
+                ddbc_error="AsyncCursor.execute called after close",
+            )
+        if not isinstance(operation, str) or not operation:
+            raise ValueError("operation must be a non-empty str")
+        await self._core.execute_async(operation, timeout_sec)
+        return self
+
+    async def fetchone(self) -> Optional[tuple]:
+        """Return the next row as a tuple, or None when the result set is exhausted."""
+        if self._closed:
+            raise InterfaceError(
+                driver_error="Cursor is closed",
+                ddbc_error="AsyncCursor.fetchone called after close",
+            )
+        return await self._core.fetchone_async()
+
+    def cancel(self) -> None:
+        """Dispatch a TDS attention to abort the currently in-flight operation.
+
+        Non-blocking. The awaiting coroutine resumes with an exception when
+        the server acknowledges the cancel.
+        """
+        if not self._closed:
+            self._core.cancel()
+
+    async def close(self) -> None:
+        """Idempotent close. Cancels any in-flight operation."""
+        if not self._closed:
+            self._core.close()
+            self._closed = True
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    async def __aenter__(self) -> "AsyncCursor":
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        await self.close()
+
+
+__all__ = ["AsyncCursor"]

pytest.ini

@@ -3,6 +3,11 @@
 markers =
     stress: marks tests as stress tests (long-running, resource-intensive)
     slow: marks tests as extra-slow (sustained load, multi-minute duration)
+    asyncio: async test cases driven by pytest-asyncio
+
+# Phase 5: opt-in async tests via @pytest.mark.asyncio (module-level or
+# per-test). Strict mode keeps existing sync suites unaffected.
+asyncio_mode = strict
 
 # Default options applied to all pytest runs
 # Default: pytest -v → Skips stress tests (fast)

tests/conftest.py

@@ -14,6 +14,11 @@
 from mssql_python import connect
 import time
 
+# Phase 5: shared fixtures for the async POC suite. Kept in this top-level
+# conftest so pytest auto-loads them without needing tests/ to be a package.
+import pytest_asyncio
+import mssql_python as _mssql_python
+
 
 def is_qemu_emulated():
     """Detect if running under QEMU user-mode emulation (e.g. ARM64 on x86_64 host).
@@ -76,3 +81,38 @@ def cursor(db_connection):
     cursor = db_connection.cursor()
     yield cursor
     cursor.close()
+
+
+# ---------------------------------------------------------------------------
+# Async POC fixtures (used by tests/test_030_async_poc.py)
+# ---------------------------------------------------------------------------
+@pytest.fixture(scope="session")
+def async_conn_str(conn_str):
+    """Session-scoped connection string. Skips the whole async suite if the
+    ``DB_CONNECTION_STRING`` env var is unset."""
+    if not conn_str:
+        pytest.skip("DB_CONNECTION_STRING is not set")
+    return conn_str
+
+
+@pytest_asyncio.fixture
+async def async_connection(async_conn_str):
+    """Function-scoped ``AsyncConnection`` — each test gets a fresh session so
+    cancellation / close tests don't leak state between cases."""
+    conn = await _mssql_python.connect_async(async_conn_str)
+    try:
+        yield conn
+    finally:
+        if not conn.closed:
+            await conn.close()
+
+
+@pytest_asyncio.fixture
+async def async_cursor(async_connection):
+    """Function-scoped ``AsyncCursor`` bound to :func:`async_connection`."""
+    cur = async_connection.cursor()
+    try:
+        yield cur
+    finally:
+        if not cur.closed:
+            await cur.close()