Remove on_start_query_execution callback from all aio cursors

Native asyncio cursors provide direct control flow via await, making
the on_start_query_execution callback unnecessary. This aligns with
AsyncCursor which also omits this callback.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: laughingman7743

pyathena/aio/arrow/cursor.py

@@ -3,7 +3,7 @@
 
 import asyncio
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
 
 from pyathena.aio.common import WithAsyncFetch
 from pyathena.arrow.converter import (
@@ -49,7 +49,6 @@ def __init__(
         unload: bool = False,
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         connect_timeout: Optional[float] = None,
         request_timeout: Optional[float] = None,
         **kwargs,
@@ -65,7 +64,6 @@ def __init__(
             kill_on_interrupt=kill_on_interrupt,
             result_reuse_enable=result_reuse_enable,
             result_reuse_minutes=result_reuse_minutes,
-            on_start_query_execution=on_start_query_execution,
             **kwargs,
         )
         self._unload = unload
@@ -92,7 +90,6 @@ async def execute(  # type: ignore[override]
         result_reuse_enable: Optional[bool] = None,
         result_reuse_minutes: Optional[int] = None,
         paramstyle: Optional[str] = None,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> "AioArrowCursor":
         """Execute a SQL query asynchronously and return results as Arrow Tables.
@@ -107,7 +104,6 @@ async def execute(  # type: ignore[override]
             result_reuse_enable: Enable Athena result reuse for this query.
             result_reuse_minutes: Minutes to reuse cached results.
             paramstyle: Parameter style ('qmark' or 'pyformat').
-            on_start_query_execution: Callback called when query starts.
             **kwargs: Additional execution parameters.
 
         Returns:
@@ -138,10 +134,6 @@ async def execute(  # type: ignore[override]
             paramstyle=paramstyle,
         )
 
-        if self._on_start_query_execution:
-            self._on_start_query_execution(self.query_id)
-        if on_start_query_execution:
-            on_start_query_execution(self.query_id)
         query_execution = await self._poll(self.query_id)
         if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
             self.result_set = await asyncio.to_thread(

pyathena/aio/common.py

@@ -365,7 +365,6 @@ def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
         self._query_id: Optional[str] = None
         self._result_set: Optional[AthenaResultSet] = None
-        self._on_start_query_execution = kwargs.get("on_start_query_execution")
 
     @property
     def arraysize(self) -> int:

pyathena/aio/cursor.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Callable, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, Union, cast
 
 from pyathena.aio.common import WithAsyncFetch
 from pyathena.aio.result_set import AthenaAioDictResultSet, AthenaAioResultSet
@@ -39,7 +39,6 @@ def __init__(
         kill_on_interrupt: bool = True,
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> None:
         super().__init__(
@@ -53,7 +52,6 @@ def __init__(
             kill_on_interrupt=kill_on_interrupt,
             result_reuse_enable=result_reuse_enable,
             result_reuse_minutes=result_reuse_minutes,
-            on_start_query_execution=on_start_query_execution,
             **kwargs,
         )
         self._result_set: Optional[AthenaAioResultSet] = None
@@ -82,7 +80,6 @@ async def execute(  # type: ignore[override]
         result_reuse_enable: Optional[bool] = None,
         result_reuse_minutes: Optional[int] = None,
         paramstyle: Optional[str] = None,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> "AioCursor":
         """Execute a SQL query asynchronously.
@@ -97,7 +94,6 @@ async def execute(  # type: ignore[override]
             result_reuse_enable: Enable result reuse (optional).
             result_reuse_minutes: Result reuse duration in minutes (optional).
             paramstyle: Parameter style to use (optional).
-            on_start_query_execution: Callback invoked with query_id after submission.
             **kwargs: Additional execution parameters.
 
         Returns:
@@ -116,11 +112,6 @@ async def execute(  # type: ignore[override]
             paramstyle=paramstyle,
         )
 
-        if self._on_start_query_execution:
-            self._on_start_query_execution(self.query_id)
-        if on_start_query_execution:
-            on_start_query_execution(self.query_id)
-
         query_execution = await self._poll(self.query_id)
         if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
             self.result_set = await self._result_set_class.create(

pyathena/aio/pandas/cursor.py

@@ -7,7 +7,6 @@
 from typing import (
     TYPE_CHECKING,
     Any,
-    Callable,
     Dict,
     Iterable,
     List,
@@ -65,7 +64,6 @@ def __init__(
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
         auto_optimize_chunksize: bool = False,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> None:
         super().__init__(
@@ -79,7 +77,6 @@ def __init__(
             kill_on_interrupt=kill_on_interrupt,
             result_reuse_enable=result_reuse_enable,
             result_reuse_minutes=result_reuse_minutes,
-            on_start_query_execution=on_start_query_execution,
             **kwargs,
         )
         self._unload = unload
@@ -113,7 +110,6 @@ async def execute(  # type: ignore[override]
         keep_default_na: bool = False,
         na_values: Optional[Iterable[str]] = ("",),
         quoting: int = 1,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> "AioPandasCursor":
         """Execute a SQL query asynchronously and return results as pandas DataFrames.
@@ -131,7 +127,6 @@ async def execute(  # type: ignore[override]
             keep_default_na: Whether to keep default pandas NA values.
             na_values: Additional values to treat as NA.
             quoting: CSV quoting behavior (pandas csv.QUOTE_* constants).
-            on_start_query_execution: Callback called when query starts.
             **kwargs: Additional pandas read_csv/read_parquet parameters.
 
         Returns:
@@ -162,10 +157,6 @@ async def execute(  # type: ignore[override]
             paramstyle=paramstyle,
         )
 
-        if self._on_start_query_execution:
-            self._on_start_query_execution(self.query_id)
-        if on_start_query_execution:
-            on_start_query_execution(self.query_id)
         query_execution = await self._poll(self.query_id)
         if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
             self.result_set = await asyncio.to_thread(

pyathena/aio/polars/cursor.py

@@ -4,7 +4,7 @@
 import asyncio
 import logging
 from multiprocessing import cpu_count
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
 
 from pyathena.aio.common import WithAsyncFetch
 from pyathena.common import CursorIterator
@@ -50,7 +50,6 @@ def __init__(
         unload: bool = False,
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         block_size: Optional[int] = None,
         cache_type: Optional[str] = None,
         max_workers: int = (cpu_count() or 1) * 5,
@@ -68,7 +67,6 @@ def __init__(
             kill_on_interrupt=kill_on_interrupt,
             result_reuse_enable=result_reuse_enable,
             result_reuse_minutes=result_reuse_minutes,
-            on_start_query_execution=on_start_query_execution,
             **kwargs,
         )
         self._unload = unload
@@ -97,7 +95,6 @@ async def execute(  # type: ignore[override]
         result_reuse_enable: Optional[bool] = None,
         result_reuse_minutes: Optional[int] = None,
         paramstyle: Optional[str] = None,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> "AioPolarsCursor":
         """Execute a SQL query asynchronously and return results as Polars DataFrames.
@@ -112,7 +109,6 @@ async def execute(  # type: ignore[override]
             result_reuse_enable: Enable Athena result reuse for this query.
             result_reuse_minutes: Minutes to reuse cached results.
             paramstyle: Parameter style ('qmark' or 'pyformat').
-            on_start_query_execution: Callback called when query starts.
             **kwargs: Additional execution parameters passed to Polars read functions.
 
         Returns:
@@ -143,10 +139,6 @@ async def execute(  # type: ignore[override]
             paramstyle=paramstyle,
         )
 
-        if self._on_start_query_execution:
-            self._on_start_query_execution(self.query_id)
-        if on_start_query_execution:
-            on_start_query_execution(self.query_id)
         query_execution = await self._poll(self.query_id)
         if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
             self.result_set = await asyncio.to_thread(

pyathena/aio/s3fs/cursor.py

@@ -3,7 +3,7 @@
 
 import asyncio
 import logging
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
 from pyathena.aio.common import WithAsyncFetch
 from pyathena.common import CursorIterator
@@ -41,7 +41,6 @@ def __init__(
         kill_on_interrupt: bool = True,
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         csv_reader: Optional[CSVReaderType] = None,
         **kwargs,
     ) -> None:
@@ -56,7 +55,6 @@ def __init__(
             kill_on_interrupt=kill_on_interrupt,
             result_reuse_enable=result_reuse_enable,
             result_reuse_minutes=result_reuse_minutes,
-            on_start_query_execution=on_start_query_execution,
             **kwargs,
         )
         self._csv_reader = csv_reader
@@ -87,7 +85,6 @@ async def execute(  # type: ignore[override]
         result_reuse_enable: Optional[bool] = None,
         result_reuse_minutes: Optional[int] = None,
         paramstyle: Optional[str] = None,
-        on_start_query_execution: Optional[Callable[[str], None]] = None,
         **kwargs,
     ) -> "AioS3FSCursor":
         """Execute a SQL query asynchronously via S3FileSystem CSV reader.
@@ -102,7 +99,6 @@ async def execute(  # type: ignore[override]
             result_reuse_enable: Enable Athena result reuse for this query.
             result_reuse_minutes: Minutes to reuse cached results.
             paramstyle: Parameter style ('qmark' or 'pyformat').
-            on_start_query_execution: Callback called when query starts.
             **kwargs: Additional execution parameters.
 
         Returns:
@@ -121,11 +117,6 @@ async def execute(  # type: ignore[override]
             paramstyle=paramstyle,
         )
 
-        if self._on_start_query_execution:
-            self._on_start_query_execution(self.query_id)
-        if on_start_query_execution:
-            on_start_query_execution(self.query_id)
-
         query_execution = await self._poll(self.query_id)
         if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
             self.result_set = await asyncio.to_thread(

tests/pyathena/aio/test_cursor.py

@@ -146,21 +146,6 @@ async def test_executemany_fetch(self, aio_cursor):
         with pytest.raises(ProgrammingError):
             await aio_cursor.fetchone()
 
-    async def test_execute_with_callback(self, aio_cursor):
-        callback_results = []
-
-        def on_start(query_id):
-            assert query_id is not None
-            assert len(query_id) > 0
-            callback_results.append(query_id)
-
-        result = await aio_cursor.execute("SELECT 1", on_start_query_execution=on_start)
-
-        assert len(callback_results) == 1
-        assert callback_results[0] == aio_cursor.query_id
-        assert result is aio_cursor
-        assert await aio_cursor.fetchone() == (1,)
-
     async def test_context_manager(self):
         conn = await _aio_connect(schema_name=ENV.schema)
         try: