PYCBC-1779: Remove asyncio.QueueEmpty exception blocks

Changes
-------
* Removed dead `except asyncio.QueueEmpty` branch from `stream_anext` and
  `AsyncRangeScanRequest.__anext__` (left over from PYCBC-1471 work)
- Dropped `op_name` param from `stream_anext`
- Removed `test_queue_empty_converted_with_op_name` from
  streaming_helpers_t.py

Change-Id: Idb3c0414d12da83005c702569609d5496f511a68
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/246875
Reviewed-by: Anirudh Lakhotia <anirudh.lakhotia@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: thejcfactor

acouchbase/analytics.py

@@ -1,4 +1,4 @@
-#  Copyright 2016-2022. Couchbase, Inc.
+#  Copyright 2016-2026. Couchbase, Inc.
 #  All Rights Reserved.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License")
@@ -124,4 +124,4 @@ def _finalize(self, exc_val=None):
         self._shutdown_executor()
 
     async def __anext__(self):
-        return await stream_anext(self, 'Analytics')
+        return await stream_anext(self)

acouchbase/kv_range_scan.py

@@ -1,17 +1,17 @@
-# Copyright 2021, Couchbase, Inc.
-# All Rights Reserved
+#  Copyright 2016-2026. Couchbase, Inc.
+#  All Rights Reserved.
 #
-# Licensed under the Apache License, Version 2.0 (the "License")
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
+#  Licensed under the Apache License, Version 2.0 (the "License")
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
 #
-#     http://www.apache.org/licenses/LICENSE-2.0
+#      http://www.apache.org/licenses/LICENSE-2.0
 #
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
 
 from __future__ import annotations
 
@@ -80,11 +80,6 @@ async def __anext__(self):
         try:
             return await self._loop.run_in_executor(self._tp_executor, self._get_next_row)
         # We can stop iterator when we receive RangeScanCompletedException
-        except asyncio.QueueEmpty:
-            exc_cls = PYCBC_ERROR_MAP.get(ExceptionMap.InternalSDKException.value, CouchbaseException)
-            excptn = exc_cls('Unexpected QueueEmpty exception caught when doing N1QL query.')
-            self._abort()
-            raise excptn
         except RangeScanCompletedException:
             self._done_streaming = True
             self._shutdown_executor()

acouchbase/n1ql.py

@@ -1,4 +1,4 @@
-#  Copyright 2016-2022. Couchbase, Inc.
+#  Copyright 2016-2026. Couchbase, Inc.
 #  All Rights Reserved.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License")
@@ -126,4 +126,4 @@ def _finalize(self, exc_val=None):
         self._shutdown_executor()
 
     async def __anext__(self):
-        return await stream_anext(self, 'N1QL')
+        return await stream_anext(self)

acouchbase/search.py

@@ -1,4 +1,4 @@
-#  Copyright 2016-2022. Couchbase, Inc.
+#  Copyright 2016-2026. Couchbase, Inc.
 #  All Rights Reserved.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License")
@@ -122,4 +122,4 @@ def _finalize(self, exc_val=None):
         self._shutdown_executor()
 
     async def __anext__(self):
-        return await stream_anext(self, 'Search')
+        return await stream_anext(self)

acouchbase/tests/streaming_helpers_t.py

@@ -84,7 +84,6 @@ class StreamingAnextTestSuite:
         'test_stop_async_iteration_finalizes_and_reads_metadata',
         'test_couchbase_exception_propagates_unchanged',
         'test_generic_exception_converted_to_internal',
-        'test_queue_empty_converted_with_op_name',
         'test_keyboard_interrupt_propagates_unconverted',
         'test_cancellation_finalizes_and_propagates',
         'test_finalize_cancels_streaming_result_on_error',
@@ -95,7 +94,7 @@ class StreamingAnextTestSuite:
     async def test_returns_row_and_ends_span(self):
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), rows=['row-1'])
         try:
-            row = await stream_anext(req, 'N1QL')
+            row = await stream_anext(req)
             assert row == 'row-1'
             # span ended on the (first) row, no error, and the op is NOT finalized mid-stream
             assert req.process_core_span_calls == [None]
@@ -108,7 +107,7 @@ async def test_returns_row_and_ends_span(self):
     async def test_stop_async_iteration_finalizes_and_reads_metadata(self):
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), rows=[])
         with pytest.raises(StopAsyncIteration):
-            await stream_anext(req, 'N1QL')
+            await stream_anext(req)
         assert req._done_streaming is True
         assert req.finalize_calls == [None]
         assert req.get_metadata_calls == 1
@@ -119,7 +118,7 @@ async def test_couchbase_exception_propagates_unchanged(self):
         err = CouchbaseException('boom')
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), raise_exc=err)
         with pytest.raises(CouchbaseException) as exc_info:
-            await stream_anext(req, 'N1QL')
+            await stream_anext(req)
         assert exc_info.value is err
         assert req.finalize_calls == [err]
         assert req.executor_shutdown is True
@@ -128,25 +127,16 @@ async def test_couchbase_exception_propagates_unchanged(self):
     async def test_generic_exception_converted_to_internal(self):
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), raise_exc=ValueError('bad'))
         with pytest.raises(InternalSDKException):
-            await stream_anext(req, 'N1QL')
+            await stream_anext(req)
         assert len(req.finalize_calls) == 1
         assert isinstance(req.finalize_calls[0], InternalSDKException)
         assert req.executor_shutdown is True
 
-    @pytest.mark.asyncio
-    async def test_queue_empty_converted_with_op_name(self):
-        req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), raise_exc=asyncio.QueueEmpty())
-        with pytest.raises(InternalSDKException) as exc_info:
-            await stream_anext(req, 'Analytics')
-        assert 'Analytics' in str(exc_info.value)
-        assert len(req.finalize_calls) == 1
-        assert req.executor_shutdown is True
-
     @pytest.mark.asyncio
     async def test_keyboard_interrupt_propagates_unconverted(self):
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), raise_exc=KeyboardInterrupt())
         with pytest.raises(KeyboardInterrupt):
-            await stream_anext(req, 'N1QL')
+            await stream_anext(req)
         # BaseException must be finalized for cleanup but NOT converted to a CouchbaseException
         assert len(req.finalize_calls) == 1
         assert isinstance(req.finalize_calls[0], KeyboardInterrupt)
@@ -157,7 +147,7 @@ async def test_cancellation_finalizes_and_propagates(self):
         # The core scenario: a task cancelled while blocked fetching the next row.
         block = threading.Event()
         req = _FakeAsyncStreamingRequest(asyncio.get_running_loop(), rows=['late-row'], block_event=block)
-        task = asyncio.ensure_future(stream_anext(req, 'N1QL'))
+        task = asyncio.ensure_future(stream_anext(req))
         # let the task reach the run_in_executor await (worker is now blocked on the event)
         await asyncio.sleep(0.05)
         task.cancel()

acouchbase/views.py

@@ -1,4 +1,4 @@
-#  Copyright 2016-2022. Couchbase, Inc.
+#  Copyright 2016-2026. Couchbase, Inc.
 #  All Rights Reserved.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License")
@@ -130,4 +130,4 @@ def _finalize(self, exc_val=None):
         self._shutdown_executor()
 
     async def __anext__(self):
-        return await stream_anext(self, 'View')
+        return await stream_anext(self)

couchbase/logic/streaming.py

@@ -1,4 +1,4 @@
-#  Copyright 2016-2022. Couchbase, Inc.
+#  Copyright 2016-2026. Couchbase, Inc.
 #  All Rights Reserved.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License")
@@ -13,8 +13,6 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-import asyncio
-
 from couchbase.exceptions import (PYCBC_ERROR_MAP,
                                   CouchbaseException,
                                   ExceptionMap)
@@ -44,12 +42,11 @@ def _internal_exception(message):
     return exc_cls(message)
 
 
-async def stream_anext(req, op_name):
+async def stream_anext(req):
     """
     **INTERNAL**
 
     Drive a single async streaming iteration for ``req`` and apply uniform teardown/error handling.
-    ``op_name`` is the human-readable operation name used in diagnostic messages (e.g. ``'N1QL'``).
     """
     try:
         # this is a blocking operation, so it is offloaded to the request's executor
@@ -60,10 +57,6 @@ async def stream_anext(req, op_name):
         # work for the first call).
         req._process_core_span()
         return row
-    except asyncio.QueueEmpty:
-        excptn = _internal_exception(f'Unexpected QueueEmpty exception caught when doing {op_name} query.')
-        req._finalize(exc_val=excptn)
-        raise excptn
     except StopAsyncIteration:
         req._done_streaming = True
         req._finalize()