From cde954cb1118b626d20dc576f11e39298b35ccdf Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 10:56:57 +0000 Subject: [PATCH 1/6] =?UTF-8?q?Phase=204=20=E2=80=94=20Python=20AsyncConne?= =?UTF-8?q?ction=20/=20AsyncCursor=20wrappers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds mssql_python/aio.py exposing AsyncConnection, AsyncCursor and the awaitable factory connect_async. Thin wrappers over the Rust mssql_py_core.PyCoreAsyncConnection / PyCoreAsyncCursor. - Every awaitable method is defined as 'async def foo(...): return await self._core.foo_async(...)' so callers can use asyncio.create_task / gather / wait_for on them. Rust future_into_py returns an awaitable (not a coroutine); create_task() would otherwise TypeError. - AsyncCursor.execute takes (operation, timeout_sec=30) and returns self. - AsyncCursor.fetchone returns tuple or None. - AsyncCursor.cancel is sync and safe to call from another task. - AsyncConnection.connect classmethod + module-level connect_async factory keep the async idiom even though connect is currently sync under the hood. - Both classes support async context managers (__aenter__ / __aexit__) and expose .closed. Idempotent close(). - Use-after-close raises InterfaceError with a clean driver/DDBC message. Wired up in mssql_python/__init__.py so users get 'mssql_python.connect_async', 'mssql_python.AsyncConnection', and 'mssql_python.AsyncCursor' via the top-level package. End-to-end wrapper smoke against local SQL Server 2022 container: - basic: SELECT 42, multi-col decode, exhaustion, re-execute - non-blocking: 60 ticks during 400 ms WAITFOR (elapsed 0.412s) - concurrent: 2x500ms queries via asyncio.gather in 0.542s - cancel: WAITFOR '00:00:10' cancelled via asyncio.create_task, exception in 3 ms - after-close: execute/fetchone raise InterfaceError Regression: tests/test_019_bulkcopy.py still 10/10 passing. --- mssql_python/__init__.py | 3 + mssql_python/aio.py | 177 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 mssql_python/aio.py diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index 29fc3a5e..e8c75984 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -60,6 +60,9 @@ # Cursor Objects from .cursor import Cursor +# Async POC surface (execute_async + fetchone_async only) +from .aio import AsyncConnection, AsyncCursor, connect_async + # Row Objects from .row import Row diff --git a/mssql_python/aio.py b/mssql_python/aio.py new file mode 100644 index 00000000..fa5a2fe1 --- /dev/null +++ b/mssql_python/aio.py @@ -0,0 +1,177 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. + +Async surface for the mssql-py-core backend. + +Exposes :class:`AsyncConnection` and :class:`AsyncCursor` alongside the module +factory :func:`connect_async`. These are thin wrappers over the Rust +``mssql_py_core.PyCoreAsyncConnection`` / ``PyCoreAsyncCursor`` classes. + +POC scope (Phase 4) — only two DB operations are exposed asynchronously: + +* :meth:`AsyncCursor.execute` — an awaitable form of ``execute``. +* :meth:`AsyncCursor.fetchone` — an awaitable form of ``fetchone``. + +Everything else (parameterised execution, ``fetchmany``/``fetchall``, async +connect, async pool, bulkcopy) is intentionally out of scope for the POC. + +Design notes +------------ +* The Rust ``execute_async`` / ``fetchone_async`` methods return a *PyO3 + awaitable*, not a Python coroutine. ``asyncio.create_task(f)`` requires a + coroutine, so every wrapper method here is defined as ``async def foo()`` + and simply ``return await self._core.foo_async(...)``. This lets callers + freely use ``create_task``, ``gather``, ``wait_for``, etc. + +* Cancellation is *cooperative*. Call :meth:`AsyncCursor.cancel` from another + task to send a TDS attention to the server; the awaited coroutine will then + resume with an exception mapped from the Rust ``OperationCancelledError``. + +* 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>``. + For true parallelism use multiple :class:`AsyncConnection` instances. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import mssql_py_core + +from mssql_python.connection_string_parser import _ConnectionStringParser +from mssql_python.exceptions import InterfaceError +from mssql_python.helpers import connstr_to_pycore_params + + +class AsyncCursor: + """Awaitable cursor. + + Not thread-safe — bind one to each asyncio task. Instances are created + via :meth:`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() + + +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", "AsyncCursor", "connect_async"] From e62eec9be868bc22c55380ce48760b75c3e45b63 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 11:03:18 +0000 Subject: [PATCH 2/6] =?UTF-8?q?Phase=205=20=E2=80=94=20pytest=20suite=20fo?= =?UTF-8?q?r=20the=20async=20POC=20(test=5F030=5Fasync=5Fpoc.py)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pytest.ini: register the 'asyncio' marker and set asyncio_mode=strict so each async test opts in explicitly (via module-level pytestmark). This leaves the 678-test sync suite unchanged. - tests/conftest.py: add async fixtures (async_conn_str session-scoped, async_connection + async_cursor function-scoped) inline so pytest auto-loads them; skip the async suite if DB_CONNECTION_STRING is unset. Function scope keeps cancellation / close tests independent. - tests/test_030_async_poc.py: 13 tests across 4 classes exercising the full POC surface end-to-end: * TestAsyncBasics (6): literal, multi-col, exhaustion, multi-row, re-execute, and async context-manager idiom via connect_async. * TestAsyncErrors (4): bad SQL, execute/fetchone after cursor close, cursor() after connection close. * TestAsyncNonBlocking (2): ticker keeps ticking during a 400 ms WAITFOR (proves loop non-blocking); two connections × 500 ms in < 0.9 s via asyncio.gather (proves multi-thread runtime). * TestAsyncCancel (1): asyncio.create_task + cur.cancel() aborts a 10 s WAITFOR in < 5 s. Results: - tests/test_030_async_poc.py: 13 passed in 1.54 s. - Regression tests/test_019_bulkcopy.py + tests/test_003_connection.py + tests/test_004_cursor.py: 678 passed / 5 skipped / 4 warnings in 36 s. - Combined pytest tests/test_019_bulkcopy.py tests/test_030_async_poc.py: 23 passed in 1.99 s (async + sync cohabit cleanly). Note: the earlier attempt at 'pytest_plugins = tests.conftest_async' failed because tests/ has no __init__.py. Folded fixtures into tests/conftest.py directly — simplest and idiomatic. --- pytest.ini | 5 + tests/conftest.py | 40 ++++++++ tests/test_030_async_poc.py | 191 ++++++++++++++++++++++++++++++++++++ 3 files changed, 236 insertions(+) create mode 100644 tests/test_030_async_poc.py diff --git a/pytest.ini b/pytest.ini index 55827f1f..b2e9d10b 100644 --- a/pytest.ini +++ b/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) diff --git a/tests/conftest.py b/tests/conftest.py index 3440e598..b36960ff 100644 --- a/tests/conftest.py +++ b/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() diff --git a/tests/test_030_async_poc.py b/tests/test_030_async_poc.py new file mode 100644 index 00000000..9fddc589 --- /dev/null +++ b/tests/test_030_async_poc.py @@ -0,0 +1,191 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. + +Phase-5 end-to-end tests for the mssql-python async POC. + +Only the two POC methods are exercised: + +* :meth:`mssql_python.AsyncCursor.execute` +* :meth:`mssql_python.AsyncCursor.fetchone` + +plus :meth:`mssql_python.AsyncCursor.cancel` for the cancellation contract. +""" + +from __future__ import annotations + +import asyncio +import os +import time + +import pytest + +import mssql_python + +# ``asyncio_mode = strict`` in pytest.ini means each async test needs the +# marker — set it once at module scope. +pytestmark = pytest.mark.asyncio + + +# --------------------------------------------------------------------------- +# TestAsyncBasics — happy-path round-trips +# --------------------------------------------------------------------------- +class TestAsyncBasics: + async def test_select_literal_returns_tuple(self, async_cursor): + r = await async_cursor.execute("SELECT 42") + # execute returns self so callers can chain fluently. + assert r is async_cursor + row = await async_cursor.fetchone() + assert row == (42,) + + async def test_select_multiple_columns(self, async_cursor): + await async_cursor.execute( + "SELECT 1, N'hi', CAST(3.14 AS FLOAT)" + ) + row = await async_cursor.fetchone() + assert row is not None + assert row[0] == 1 + assert row[1] == "hi" + assert abs(row[2] - 3.14) < 1e-9 + + async def test_fetchone_returns_none_after_exhausting(self, async_cursor): + await async_cursor.execute("SELECT 1") + assert (await async_cursor.fetchone()) == (1,) + assert (await async_cursor.fetchone()) is None + # Idempotent: a second post-exhaustion fetch is still None. + assert (await async_cursor.fetchone()) is None + + async def test_multiple_rows_via_repeated_fetchone(self, async_cursor): + await async_cursor.execute( + "SELECT n FROM (VALUES (1),(2),(3)) AS t(n) ORDER BY n" + ) + rows = [] + while (r := await async_cursor.fetchone()) is not None: + rows.append(r) + assert rows == [(1,), (2,), (3,)] + + async def test_reexecute_after_drain(self, async_cursor): + await async_cursor.execute("SELECT 10") + assert (await async_cursor.fetchone()) == (10,) + assert (await async_cursor.fetchone()) is None + await async_cursor.execute("SELECT 20") + assert (await async_cursor.fetchone()) == (20,) + + async def test_context_manager(self, async_conn_str): + async with await mssql_python.connect_async(async_conn_str) as conn: + assert not conn.closed + async with conn.cursor() as cur: + assert not cur.closed + await cur.execute("SELECT 7") + assert (await cur.fetchone()) == (7,) + assert cur.closed + assert conn.closed + + +# --------------------------------------------------------------------------- +# TestAsyncErrors — error path fidelity +# --------------------------------------------------------------------------- +class TestAsyncErrors: + async def test_bad_sql_raises(self, async_cursor): + with pytest.raises(Exception): + await async_cursor.execute("SELECT * FROM __no_such_table__xyz") + + async def test_execute_after_cursor_close_raises(self, async_cursor): + await async_cursor.close() + with pytest.raises(Exception): + await async_cursor.execute("SELECT 1") + + async def test_fetchone_after_cursor_close_raises(self, async_cursor): + await async_cursor.close() + with pytest.raises(Exception): + await async_cursor.fetchone() + + async def test_cursor_from_closed_connection_raises(self, async_connection): + await async_connection.close() + with pytest.raises(Exception): + async_connection.cursor() + + +# --------------------------------------------------------------------------- +# TestAsyncNonBlocking — proves the event loop is not blocked +# --------------------------------------------------------------------------- +class TestAsyncNonBlocking: + async def test_event_loop_remains_responsive(self, async_cursor): + """A ticker task must make progress while a server-side WAITFOR runs.""" + ticks = 0 + + async def ticker(): + nonlocal ticks + # 40 iterations * 10ms = ~400ms max ticker duration, mirroring + # the server WAITFOR budget below. + for _ in range(40): + await asyncio.sleep(0.01) + ticks += 1 + + t = asyncio.create_task(ticker()) + start = time.monotonic() + await async_cursor.execute("WAITFOR DELAY '00:00:00.400'; SELECT 1") + row = await async_cursor.fetchone() + elapsed = time.monotonic() - start + await t + + assert row == (1,) + # A properly non-blocking driver lets the loop dispatch the ticker + # ~40 times during a 400ms server wait. Allow half of that as slack + # for slow CI runners. + assert ticks >= 20, ( + f"event loop appears blocked " + f"(ticks={ticks}, elapsed={elapsed:.3f}s)" + ) + + async def test_two_connections_run_concurrently(self, async_conn_str): + """asyncio.gather across independent connections must overlap.""" + + async def one(): + async with await mssql_python.connect_async(async_conn_str) as conn: + async with conn.cursor() as cur: + await cur.execute( + "WAITFOR DELAY '00:00:00.500'; SELECT 1" + ) + return await cur.fetchone() + + start = time.monotonic() + results = await asyncio.gather(one(), one()) + elapsed = time.monotonic() - start + + assert results == [(1,), (1,)] + # Two 500ms queries would take ~1s if serialised on a single-thread + # runtime. Multi-thread + parallel connections should finish well + # under that. + assert elapsed < 0.9, ( + f"queries appear serialised (elapsed={elapsed:.3f}s)" + ) + + +# --------------------------------------------------------------------------- +# TestAsyncCancel — TDS attention plumbing +# --------------------------------------------------------------------------- +class TestAsyncCancel: + async def test_explicit_cancel_aborts_long_query(self, async_connection): + cur = async_connection.cursor() + try: + task = asyncio.create_task( + cur.execute("WAITFOR DELAY '00:00:10'", timeout_sec=30) + ) + # Let the batch reach the server and start waiting. + await asyncio.sleep(0.2) + + start = time.monotonic() + cur.cancel() + + with pytest.raises(Exception): + await asyncio.wait_for(task, timeout=5) + elapsed = time.monotonic() - start + + # The server acknowledges attention within a few ms once it's + # in a wait; leave generous headroom for slow CI runners. + assert elapsed < 5, ( + f"cancel took too long ({elapsed:.3f}s)" + ) + finally: + await cur.close() From f9eab1e93c4bb791c74d85c1ec764a488996eb70 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 11:16:34 +0000 Subject: [PATCH 3/6] chore: ignore docs/async/ (local-only async POC design docs) Keeps ASYNC_POC_SPEC.md / ASYNC_POC_RUNBOOK.md as personal working notes without committing them upstream. History for the docs commit (previously 9fe5fe60) has been rebase-dropped from this branch. --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index 080fc9e5..84d38fc5 100644 --- a/.gitignore +++ b/.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/ From adc65b239a30cd87e5be9ebb21bb4fbe0cbaa7d5 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 11:46:17 +0000 Subject: [PATCH 4/6] refactor: split aio.py into cursor_async.py + connection_async.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors the sync pair (cursor.py / connection.py) — one class per file, keeping module names discoverable by their responsibility. - mssql_python/cursor_async.py AsyncCursor - mssql_python/connection_async.py AsyncConnection + connect_async factory - mssql_python/aio.py deleted Public surface unchanged: mssql_python.AsyncConnection, AsyncCursor and connect_async are still re-exported from the package __init__ and remain the recommended entry points for user code. Verified: 23/23 async + bulkcopy tests pass after the split. --- mssql_python/__init__.py | 3 +- mssql_python/aio.py | 177 ------------------------------- mssql_python/connection_async.py | 101 ++++++++++++++++++ mssql_python/cursor_async.py | 91 ++++++++++++++++ 4 files changed, 194 insertions(+), 178 deletions(-) delete mode 100644 mssql_python/aio.py create mode 100644 mssql_python/connection_async.py create mode 100644 mssql_python/cursor_async.py diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index e8c75984..ac3bc0d1 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -61,7 +61,8 @@ from .cursor import Cursor # Async POC surface (execute_async + fetchone_async only) -from .aio import AsyncConnection, AsyncCursor, connect_async +from .connection_async import AsyncConnection, connect_async +from .cursor_async import AsyncCursor # Row Objects from .row import Row diff --git a/mssql_python/aio.py b/mssql_python/aio.py deleted file mode 100644 index fa5a2fe1..00000000 --- a/mssql_python/aio.py +++ /dev/null @@ -1,177 +0,0 @@ -""" -Copyright (c) Microsoft Corporation. -Licensed under the MIT license. - -Async surface for the mssql-py-core backend. - -Exposes :class:`AsyncConnection` and :class:`AsyncCursor` alongside the module -factory :func:`connect_async`. These are thin wrappers over the Rust -``mssql_py_core.PyCoreAsyncConnection`` / ``PyCoreAsyncCursor`` classes. - -POC scope (Phase 4) — only two DB operations are exposed asynchronously: - -* :meth:`AsyncCursor.execute` — an awaitable form of ``execute``. -* :meth:`AsyncCursor.fetchone` — an awaitable form of ``fetchone``. - -Everything else (parameterised execution, ``fetchmany``/``fetchall``, async -connect, async pool, bulkcopy) is intentionally out of scope for the POC. - -Design notes ------------- -* The Rust ``execute_async`` / ``fetchone_async`` methods return a *PyO3 - awaitable*, not a Python coroutine. ``asyncio.create_task(f)`` requires a - coroutine, so every wrapper method here is defined as ``async def foo()`` - and simply ``return await self._core.foo_async(...)``. This lets callers - freely use ``create_task``, ``gather``, ``wait_for``, etc. - -* Cancellation is *cooperative*. Call :meth:`AsyncCursor.cancel` from another - task to send a TDS attention to the server; the awaited coroutine will then - resume with an exception mapped from the Rust ``OperationCancelledError``. - -* 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>``. - For true parallelism use multiple :class:`AsyncConnection` instances. -""" - -from __future__ import annotations - -from typing import Any, Optional - -import mssql_py_core - -from mssql_python.connection_string_parser import _ConnectionStringParser -from mssql_python.exceptions import InterfaceError -from mssql_python.helpers import connstr_to_pycore_params - - -class AsyncCursor: - """Awaitable cursor. - - Not thread-safe — bind one to each asyncio task. Instances are created - via :meth:`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() - - -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", "AsyncCursor", "connect_async"] diff --git a/mssql_python/connection_async.py b/mssql_python/connection_async.py new file mode 100644 index 00000000..96ec65ae --- /dev/null +++ b/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>``. 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"] diff --git a/mssql_python/cursor_async.py b/mssql_python/cursor_async.py new file mode 100644 index 00000000..72b78953 --- /dev/null +++ b/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"] From 6c272d8b32cfec289880ec9a1177710e9e5c8279 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 11:51:20 +0000 Subject: [PATCH 5/6] test: add 100-connection scale tests for the async POC Two tests in a new TestAsyncScale class exercise the pyo3-async-runtimes + Tokio pipeline under real concurrent load: - test_hundred_connections_correctness: 100 tasks, each opening a fresh AsyncConnection, running SELECT n, fetchone, and closing. Verifies every value round-trips and the pipeline handles 100 concurrent sessions from a single Python process. - test_hundred_connections_true_parallelism: 100 tasks each running a 200 ms server-side WAITFOR. Asserts wall time is well below the 20 s serial baseline, proving the multi-thread Tokio runtime + non-blocking future_into_py futures actually run in parallel. Both marked @pytest.mark.slow (currently included by default; can be excluded via pytest -m 'not slow' if a runner is resource-constrained). Measurements against local SQL Server 2022 container: - 100 connect+select+close cycles: 0.963 s (~10 ms/conn amortised) - 100 x 200 ms WAITFOR: 1.107 s (~18x over serial baseline) Combined pytest tests/test_030_async_poc.py tests/test_019_bulkcopy.py: 25 passed in 3.51 s. --- tests/test_030_async_poc.py | 83 +++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/tests/test_030_async_poc.py b/tests/test_030_async_poc.py index 9fddc589..a63aa598 100644 --- a/tests/test_030_async_poc.py +++ b/tests/test_030_async_poc.py @@ -189,3 +189,86 @@ async def test_explicit_cancel_aborts_long_query(self, async_connection): ) finally: await cur.close() + + +# --------------------------------------------------------------------------- +# TestAsyncScale — many concurrent connections against the Rust runtime +# --------------------------------------------------------------------------- +class TestAsyncScale: + """Scale tests for the async POC — 100 concurrent connections. + + Each task opens a fresh :class:`AsyncConnection`, runs a single query + and closes. This exercises: + + * the process-wide Tokio runtime under load, + * the pyo3-async-runtimes → asyncio bridge with 100 futures in flight, + * GIL discipline — the row decode / connection lifecycle must not + serialise the event loop, + * SQL Server's ability to handle 100 concurrent sessions from a single + Python process (well within default limits, but worth exercising). + + Marked ``slow`` so ``pytest -m "not slow"`` can skip them if a runner + is memory-constrained. + """ + + @pytest.mark.slow + async def test_hundred_connections_correctness(self, async_conn_str): + """100 tasks, each picking a distinct constant. All must round-trip.""" + n_tasks = 100 + + async def one(idx: int) -> tuple: + async with await mssql_python.connect_async(async_conn_str) as conn: + async with conn.cursor() as cur: + await cur.execute(f"SELECT {idx}") + return await cur.fetchone() + + start = time.monotonic() + results = await asyncio.gather(*(one(i) for i in range(n_tasks))) + elapsed = time.monotonic() - start + + expected = [(i,) for i in range(n_tasks)] + assert results == expected, ( + f"one or more of {n_tasks} tasks returned an unexpected value" + ) + # 100 sequential connect+select+close cycles would take many seconds + # on any reasonable runner; the parallel run should finish comfortably + # inside a generous ceiling. This is a scalability sanity check — + # tighten with data once we have baselines. + assert elapsed < 30, ( + f"100 concurrent connections took {elapsed:.3f}s (>30s ceiling)" + ) + print(f"[scale.correctness] 100 connections in {elapsed:.3f}s") + + @pytest.mark.slow + async def test_hundred_connections_true_parallelism(self, async_conn_str): + """100 tasks each running a 200 ms server WAITFOR. + + If the pipeline truly runs in parallel, wall time is bounded by + (200 ms + connect overhead) times a small multiplier — not by the + 200 ms × 100 = 20 s serial baseline. + """ + n_tasks = 100 + + async def one() -> tuple: + async with await mssql_python.connect_async(async_conn_str) as conn: + async with conn.cursor() as cur: + await cur.execute("WAITFOR DELAY '00:00:00.200'; SELECT 1") + return await cur.fetchone() + + start = time.monotonic() + results = await asyncio.gather(*(one() for _ in range(n_tasks))) + elapsed = time.monotonic() - start + + assert results == [(1,)] * n_tasks + # Serial baseline is ~20s. Parallel target is well under that; the + # tokio multi-thread runtime + non-blocking futures should finish + # inside a few seconds. Ceiling picked to leave headroom for slow + # SQL Server startup on CI. + assert elapsed < 10, ( + f"100 x 200 ms WAITFORs took {elapsed:.3f}s " + f"(expected << 20 s serial baseline)" + ) + print( + f"[scale.parallel] 100 x 200 ms WAITFORs in {elapsed:.3f}s " + f"(serial baseline ~20s)" + ) From ab42a354d4dc01ce054d77011f6895a14a184874 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 15 Jul 2026 12:06:29 +0000 Subject: [PATCH 6/6] test: harden 100-connection scale tests with a complex, long-running query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the trivial SELECT / WAITFOR stubs with a T-SQL batch that is both non-trivial and predictably verifiable: WAITFOR DELAY ... -- keeps it long-running WITH nums AS ( recursive CTE producing 50 rows ) SELECT task_id INT, -- interpolated identity n INT, -- row order 1..50 square INT (n*n), -- integer arithmetic label NVARCHAR, -- per-row string with task_id + n root FLOAT (SQRT(n)) -- floating-point decode FROM nums OPTION (MAXRECURSION 50) Every value is server-computed, so any driver-side corruption (wrong column order, truncated string, wrong float bits, wrong integer width) surfaces as an assertion failure. Row-by-row validation across 5 000 rows per test proves the streaming fetchone_async decode is correct at scale, not just for a single sentinel value. Helpers extracted for readability: - _complex_long_running_query(task_id, wait_ms) — batch builder. - _verify_task_rows(task_id, rows) — deterministic per-row assertions. Tests updated: - test_hundred_connections_complex_query_correctness (100 ms wait) - test_hundred_connections_long_running_parallelism (200 ms wait) Measurements against local SQL Server 2022 container: - Correctness (5 000 rows verified, 100 ms wait/task): 1.782 s @ 2 806 r/s - Parallelism (5 000 rows verified, 200 ms wait/task): 1.791 s @ 2 791 r/s (serial baseline ~20 s just for the WAITFORs) Combined pytest tests/test_030_async_poc.py tests/test_019_bulkcopy.py: 25 passed in 5.17 s. --- tests/test_030_async_poc.py | 197 ++++++++++++++++++++++++++++-------- 1 file changed, 153 insertions(+), 44 deletions(-) diff --git a/tests/test_030_async_poc.py b/tests/test_030_async_poc.py index a63aa598..d9b49c19 100644 --- a/tests/test_030_async_poc.py +++ b/tests/test_030_async_poc.py @@ -194,81 +194,190 @@ async def test_explicit_cancel_aborts_long_query(self, async_connection): # --------------------------------------------------------------------------- # TestAsyncScale — many concurrent connections against the Rust runtime # --------------------------------------------------------------------------- +_ROWS_PER_TASK: int = 50 + + +def _complex_long_running_query(task_id: int, wait_ms: int) -> str: + """T-SQL that is both non-trivial and predictably verifiable per row. + + Structure: + * ``WAITFOR DELAY`` prefix — makes each batch a **long-running** + statement so the parallel wall-time assertion is meaningful. + * Recursive CTE emitting ``_ROWS_PER_TASK`` rows. + * Five computed columns exercising the row-decode pipeline: + - ``task_id`` INT — the interpolated task id (identity check). + - ``n`` INT — 1..ROWS_PER_TASK (row order). + - ``square`` INT — ``n * n`` (integer arithmetic). + - ``label`` NVARCHAR — per-row string, checked exactly. + - ``root`` FLOAT — ``SQRT(n)`` (floating-point decode). + + ``task_id`` is interpolated into the SQL text because the POC does not + yet support parameterised queries. Every value is server-computed so + driver-side tampering would surface as a mismatch. + """ + return f""" + WAITFOR DELAY '00:00:00.{wait_ms:03d}'; + WITH nums AS ( + SELECT CAST(1 AS INT) AS n + UNION ALL + SELECT n + 1 FROM nums WHERE n < {_ROWS_PER_TASK} + ) + SELECT + CAST({task_id} AS INT) AS task_id, + n, + n * n AS square, + CAST(N'task-' + CAST({task_id} AS nvarchar(8)) + + N'-row-' + CAST(n AS nvarchar(8)) + AS nvarchar(40)) AS label, + CAST(SQRT(CAST(n AS FLOAT)) AS FLOAT) AS root + FROM nums + OPTION (MAXRECURSION {_ROWS_PER_TASK}); + """ + + +def _verify_task_rows(task_id: int, rows: list) -> None: + """Row-by-row validation for :func:`_complex_long_running_query`.""" + import math + + assert len(rows) == _ROWS_PER_TASK, ( + f"task {task_id} returned {len(rows)} rows, expected {_ROWS_PER_TASK}" + ) + for expected_n, row in enumerate(rows, start=1): + assert len(row) == 5, ( + f"task {task_id} row {expected_n} has {len(row)} columns, expected 5" + ) + got_task_id, got_n, got_square, got_label, got_root = row + assert got_task_id == task_id, ( + f"task {task_id} row {expected_n}: task_id={got_task_id!r}" + ) + assert got_n == expected_n, ( + f"task {task_id}: row {expected_n} had n={got_n!r}" + ) + assert got_square == expected_n * expected_n, ( + f"task {task_id} row {expected_n}: square={got_square!r}, " + f"expected {expected_n * expected_n}" + ) + assert got_label == f"task-{task_id}-row-{expected_n}", ( + f"task {task_id} row {expected_n}: label={got_label!r}" + ) + assert isinstance(got_root, float), ( + f"task {task_id} row {expected_n}: " + f"expected FLOAT to decode as Python float, got {type(got_root).__name__}" + ) + assert abs(got_root - math.sqrt(expected_n)) < 1e-9, ( + f"task {task_id} row {expected_n}: " + f"root={got_root!r}, expected {math.sqrt(expected_n)!r}" + ) + + class TestAsyncScale: - """Scale tests for the async POC — 100 concurrent connections. + """Scale tests for the async POC — 100 concurrent connections running + a complex, long-running SELECT with full per-row data verification. - Each task opens a fresh :class:`AsyncConnection`, runs a single query - and closes. This exercises: + Each task opens a fresh :class:`AsyncConnection`, runs a query that: - * the process-wide Tokio runtime under load, - * the pyo3-async-runtimes → asyncio bridge with 100 futures in flight, - * GIL discipline — the row decode / connection lifecycle must not - serialise the event loop, - * SQL Server's ability to handle 100 concurrent sessions from a single - Python process (well within default limits, but worth exercising). + * server-waits (``WAITFOR DELAY``) — long-running behaviour, + * emits ``_ROWS_PER_TASK`` rows from a recursive CTE, + * has five computed columns of three distinct types (int, nvarchar, + float) so every row can be validated exactly. - Marked ``slow`` so ``pytest -m "not slow"`` can skip them if a runner - is memory-constrained. + Every value is server-computed, so any driver-side corruption (wrong + column order, truncated string, wrong float bits, wrong integer + width) surfaces as an assertion failure. + + Exercises simultaneously: + * the process-wide Tokio runtime under load, + * the pyo3-async-runtimes → asyncio bridge with 100 futures in flight, + * GIL discipline — row decode of 5 000 rows over 100 connections must + not serialise the event loop, + * streaming ``fetchone`` correctness across many rows per connection. + + Marked ``slow`` so ``pytest -m "not slow"`` can skip them. """ @pytest.mark.slow - async def test_hundred_connections_correctness(self, async_conn_str): - """100 tasks, each picking a distinct constant. All must round-trip.""" + async def test_hundred_connections_complex_query_correctness( + self, async_conn_str + ): + """100 tasks, complex long-running query, every row checked.""" n_tasks = 100 + # Modest wait keeps this test's wall time low while still making + # the batch "long-running" from the driver's point of view. + wait_ms = 100 - async def one(idx: int) -> tuple: + async def one(task_id: int) -> list: async with await mssql_python.connect_async(async_conn_str) as conn: async with conn.cursor() as cur: - await cur.execute(f"SELECT {idx}") - return await cur.fetchone() + await cur.execute( + _complex_long_running_query(task_id, wait_ms=wait_ms), + timeout_sec=60, + ) + rows: list = [] + while (r := await cur.fetchone()) is not None: + rows.append(r) + return rows start = time.monotonic() results = await asyncio.gather(*(one(i) for i in range(n_tasks))) elapsed = time.monotonic() - start - expected = [(i,) for i in range(n_tasks)] - assert results == expected, ( - f"one or more of {n_tasks} tasks returned an unexpected value" + # Per-task, per-row verification: server-computed values must match + # driver-decoded values exactly for every one of the 5 000 rows. + for task_id, rows in enumerate(results): + _verify_task_rows(task_id, rows) + + total_rows = n_tasks * _ROWS_PER_TASK + # 100 connections x ~100 ms wait + query = ~10 s serial baseline. + # Parallel budget must be comfortably below that. + assert elapsed < 10, ( + f"100 concurrent complex-query connections took {elapsed:.3f}s " + f"(>10s ceiling; serial baseline ~10s)" ) - # 100 sequential connect+select+close cycles would take many seconds - # on any reasonable runner; the parallel run should finish comfortably - # inside a generous ceiling. This is a scalability sanity check — - # tighten with data once we have baselines. - assert elapsed < 30, ( - f"100 concurrent connections took {elapsed:.3f}s (>30s ceiling)" + print( + f"[scale.complex-correctness] 100 conns x {_ROWS_PER_TASK} rows " + f"= {total_rows} rows in {elapsed:.3f}s " + f"({total_rows / elapsed:.0f} rows/s)" ) - print(f"[scale.correctness] 100 connections in {elapsed:.3f}s") @pytest.mark.slow - async def test_hundred_connections_true_parallelism(self, async_conn_str): - """100 tasks each running a 200 ms server WAITFOR. - - If the pipeline truly runs in parallel, wall time is bounded by - (200 ms + connect overhead) times a small multiplier — not by the - 200 ms × 100 = 20 s serial baseline. + async def test_hundred_connections_long_running_parallelism( + self, async_conn_str + ): + """100 tasks each running a 200 ms + complex CTE query. + + Serial baseline is ~20 s (100 × 200 ms just for the WAITFOR). + Parallel target is well under that. Every row is still verified — + this is not just a wall-time assertion. """ n_tasks = 100 + wait_ms = 200 - async def one() -> tuple: + async def one(task_id: int) -> list: async with await mssql_python.connect_async(async_conn_str) as conn: async with conn.cursor() as cur: - await cur.execute("WAITFOR DELAY '00:00:00.200'; SELECT 1") - return await cur.fetchone() + await cur.execute( + _complex_long_running_query(task_id, wait_ms=wait_ms), + timeout_sec=60, + ) + rows: list = [] + while (r := await cur.fetchone()) is not None: + rows.append(r) + return rows start = time.monotonic() - results = await asyncio.gather(*(one() for _ in range(n_tasks))) + results = await asyncio.gather(*(one(i) for i in range(n_tasks))) elapsed = time.monotonic() - start - assert results == [(1,)] * n_tasks - # Serial baseline is ~20s. Parallel target is well under that; the - # tokio multi-thread runtime + non-blocking futures should finish - # inside a few seconds. Ceiling picked to leave headroom for slow - # SQL Server startup on CI. + for task_id, rows in enumerate(results): + _verify_task_rows(task_id, rows) + + total_rows = n_tasks * _ROWS_PER_TASK assert elapsed < 10, ( - f"100 x 200 ms WAITFORs took {elapsed:.3f}s " + f"100 x (200 ms WAITFOR + complex CTE) took {elapsed:.3f}s " f"(expected << 20 s serial baseline)" ) print( - f"[scale.parallel] 100 x 200 ms WAITFORs in {elapsed:.3f}s " - f"(serial baseline ~20s)" + f"[scale.long-running-parallel] 100 conns x {_ROWS_PER_TASK} rows " + f"= {total_rows} rows in {elapsed:.3f}s " + f"(serial baseline ~20s, {total_rows / elapsed:.0f} rows/s)" )