diff --git a/CHANGELOG.md b/CHANGELOG.md index fc442de7..771e26ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,5 @@ +* Deprecated the table client scan query methods — `TableClient.scan_query`, `TableClient.async_scan_query` and the async `ydb.aio.TableClient.scan_query`: they now emit a `DeprecationWarning` and keep working as before, use QueryService (`ydb.QuerySessionPool` / `ydb.aio.QuerySessionPool`) instead + ## 3.31.4 ## * Fixed async `QuerySessionPool` permanently losing a pool slot when `acquire()` was cancelled while a new session was being created: `asyncio.CancelledError` no longer leaks the pool size counter, so a pool under deadline-driven cancellations can no longer end up exhausted and blocking forever. A cancelled or interrupted session attach now also closes the session instead of orphaning it server-side diff --git a/ydb/aio/table.py b/ydb/aio/table.py index f76a934b..48b3dab7 100644 --- a/ydb/aio/table.py +++ b/ydb/aio/table.py @@ -2,6 +2,7 @@ import logging import time import typing +import warnings from typing import ( Any, @@ -19,6 +20,7 @@ from ydb.table import ( BaseSession, BaseTableClient, + _SCAN_QUERY_DEPRECATION_MESSAGE, _scan_query_request_factory, _wrap_scan_query_response, BaseTxContext, @@ -176,6 +178,14 @@ async def describe_system_view(self, path, settings=None): # pylint: disable=W0 return await super().describe_system_view(path, settings) async def scan_query(self, query, parameters=None, settings=None): # pylint: disable=W0236 + """ + Deprecated: use QueryService (:class:`ydb.aio.QuerySessionPool`) instead. + """ + warnings.warn( + _SCAN_QUERY_DEPRECATION_MESSAGE.format(method="scan_query", pool="ydb.aio.QuerySessionPool"), + DeprecationWarning, + stacklevel=2, + ) request = _scan_query_request_factory(query, parameters, settings) response = await self._driver( request, diff --git a/ydb/table.py b/ydb/table.py index 8fbc5e78..e326a890 100644 --- a/ydb/table.py +++ b/ydb/table.py @@ -6,6 +6,7 @@ import logging import enum import typing +import warnings from typing import ( Any, @@ -985,6 +986,11 @@ def __init__(self, yql_text, parameters_types): self.parameters_types = parameters_types +_SCAN_QUERY_DEPRECATION_MESSAGE = ( + "{method} is deprecated and will be removed in a future release, use QueryService ({pool}) instead" +) + + def _wrap_scan_query_response(response, table_client_settings): issues._process_response(response) return ScanQueryResult(response.result, table_client_settings) @@ -1180,6 +1186,9 @@ def session(self): @abstractmethod def scan_query(self, query, parameters=None, settings=None): + """ + Deprecated: use QueryService (:class:`ydb.QuerySessionPool`) instead. + """ pass @abstractmethod @@ -1208,6 +1217,14 @@ def session(self): def scan_query(self, query, parameters=None, settings=None): # type: (ydb.ScanQuery, tuple, ydb.BaseRequestSettings) -> _utilities.SyncResponseIterator + """ + Deprecated: use QueryService (:class:`ydb.QuerySessionPool`) instead. + """ + warnings.warn( + _SCAN_QUERY_DEPRECATION_MESSAGE.format(method="scan_query", pool="ydb.QuerySessionPool"), + DeprecationWarning, + stacklevel=2, + ) request = _scan_query_request_factory(query, parameters, settings) stream_it = self._driver( request, @@ -1269,6 +1286,14 @@ def __del__(self): def async_scan_query(self, query, parameters=None, settings=None): # type: (ydb.ScanQuery, tuple, ydb.BaseRequestSettings) -> _utilities.AsyncResponseIterator + """ + Deprecated: use QueryService (:class:`ydb.QuerySessionPool`) instead. + """ + warnings.warn( + _SCAN_QUERY_DEPRECATION_MESSAGE.format(method="async_scan_query", pool="ydb.QuerySessionPool"), + DeprecationWarning, + stacklevel=2, + ) request = _scan_query_request_factory(query, parameters, settings) stream_it = self._driver( request, diff --git a/ydb/table_test.py b/ydb/table_test.py index ad158260..e5c24c38 100644 --- a/ydb/table_test.py +++ b/ydb/table_test.py @@ -3,7 +3,9 @@ import pytest from unittest import mock -from . import issues, convert, types, _apis, scheme, _session_impl +from . import issues, convert, types, _apis, scheme, _session_impl, _utilities +from .aio import _utilities as _aio_utilities +from .aio.table import TableClient as AioTableClient from .table import SystemViewSchemeEntry, TableClient from .retries import ( @@ -336,3 +338,68 @@ def future(self, request, stub, method, wrap_fn, settings, wrap_args, *rest): assert driver.request.path == "/local/.sys/partition_stats" assert isinstance(entry, SystemViewSchemeEntry) assert entry.sys_view_name == "partition_stats" + + +class _FakeScanQueryDriver: + def __call__(self, request, stub, method, settings=None): + self.request = request + self.method = method + return iter(()) + + +class _FakeAsyncScanQueryDriver: + async def __call__(self, request, stub, method, settings=None): + self.request = request + self.method = method + return _EmptyAsyncStream() + + +class _EmptyAsyncStream: + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + +def test_scan_query_warns_and_still_returns_iterator(): + driver = _FakeScanQueryDriver() + + with pytest.warns(DeprecationWarning) as record: + stream = TableClient(driver).scan_query("SELECT 1") + + assert str(record[0].message) == ( + "scan_query is deprecated and will be removed in a future release, " + "use QueryService (ydb.QuerySessionPool) instead" + ) + assert driver.method == _apis.TableService.StreamExecuteScanQuery + assert isinstance(stream, _utilities.SyncResponseIterator) + + +def test_async_scan_query_warns_and_still_returns_iterator(): + driver = _FakeScanQueryDriver() + + with pytest.warns(DeprecationWarning) as record: + stream = TableClient(driver).async_scan_query("SELECT 1") + + assert str(record[0].message) == ( + "async_scan_query is deprecated and will be removed in a future release, " + "use QueryService (ydb.QuerySessionPool) instead" + ) + assert driver.method == _apis.TableService.StreamExecuteScanQuery + assert isinstance(stream, _utilities.AsyncResponseIterator) + + +async def test_aio_scan_query_warns_and_points_to_async_pool(): + driver = _FakeAsyncScanQueryDriver() + + with pytest.warns(DeprecationWarning) as record: + stream = await AioTableClient(driver).scan_query("SELECT 1") + + # The async client must not send users to the sync session pool. + assert str(record[0].message) == ( + "scan_query is deprecated and will be removed in a future release, " + "use QueryService (ydb.aio.QuerySessionPool) instead" + ) + assert driver.method == _apis.TableService.StreamExecuteScanQuery + assert isinstance(stream, _aio_utilities.AsyncResponseIterator)