From bdb33edde581590dc0f0153832abcfb3c48b5653 Mon Sep 17 00:00:00 2001 From: Callan Gray Date: Wed, 8 Jul 2026 11:20:02 +0800 Subject: [PATCH 1/3] Lightweight queue monad Signed-off-by: Callan Gray --- src/athreading/iterator.py | 35 ++++++++++++++++++++--------------- 1 file changed, 20 insertions(+), 15 deletions(-) diff --git a/src/athreading/iterator.py b/src/athreading/iterator.py index f0ff5a4..1ab7415 100644 --- a/src/athreading/iterator.py +++ b/src/athreading/iterator.py @@ -3,14 +3,13 @@ from __future__ import annotations import asyncio -import dataclasses import functools import queue import sys import threading from collections.abc import Callable, Iterator from concurrent.futures import ThreadPoolExecutor -from typing import TYPE_CHECKING, Generic, Optional, TypeVar, Union, cast +from typing import TYPE_CHECKING, Generic, Optional, TypeVar, Union from athreading.aliases import AsyncIteratorContext @@ -26,6 +25,17 @@ _ParamsT = ParamSpec("_ParamsT") _YieldT = TypeVar("_YieldT") _T = TypeVar("_T") +_E = TypeVar("_E", bound=BaseException) + + +class _Ok(Generic[_T]): + def __init__(self, value: _T): + self.value = value + + +class _Err(Generic[_E]): + def __init__(self, error: _E): + self.error = error @overload @@ -119,12 +129,6 @@ def decorator( return decorator -@dataclasses.dataclass -class _Result(Generic[_T]): - value: _T | None - error: Exception | None - - class ThreadedAsyncIterator(AsyncIteratorContext[_YieldT]): """Wraps a synchronous iterator with an executor and exposes an AsyncIteratorContext.""" @@ -144,7 +148,9 @@ def __init__( """ self._yield_semaphore = asyncio.Semaphore(0) self._done_event = threading.Event() - self._queue: queue.Queue[_Result[_YieldT]] = queue.Queue(buffer_maxsize or 0) + self._queue: queue.Queue[_Ok[_YieldT] | _Err[Exception]] = queue.Queue( + buffer_maxsize or 0 + ) self._iterator = iterator self._executor = executor self._worker_future: Optional[asyncio.Future[None]] = None @@ -178,17 +184,16 @@ async def __anext__(self) -> _YieldT: await self._yield_semaphore.acquire() if not self._queue.empty(): result = self._queue.get(False) - if result.error is not None: + if isinstance(result, _Err): raise result.error - else: - return cast(_YieldT, result.value) + return result.value raise StopAsyncIteration def __worker_threadsafe(self) -> None: """Stream the synchronous iterator to the queue and notify the async thread.""" try: for item in self._iterator: - self._queue.put(_Result(value=item, error=None)) + self._queue.put(_Ok(item)) self._loop.call_soon_threadsafe(self._yield_semaphore.release) while self._queue.full() and not self._done_event.is_set(): @@ -197,8 +202,8 @@ def __worker_threadsafe(self) -> None: if self._done_event.is_set(): break - except Exception as e: - self._queue.put(_Result(value=None, error=e)) + except Exception as e: # noqa: BLE001 + self._queue.put(_Err(e)) self._loop.call_soon_threadsafe(self._yield_semaphore.release) finally: self._done_event.set() From e170212cdcd7df4ce89b45e05dc44e9b625bf42b Mon Sep 17 00:00:00 2001 From: Callan Gray Date: Wed, 8 Jul 2026 11:51:03 +0800 Subject: [PATCH 2/3] Simplify typevars Signed-off-by: Callan Gray --- src/athreading/iterator.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/athreading/iterator.py b/src/athreading/iterator.py index 1ab7415..a9bdd91 100644 --- a/src/athreading/iterator.py +++ b/src/athreading/iterator.py @@ -21,11 +21,9 @@ if TYPE_CHECKING: from types import TracebackType - _ParamsT = ParamSpec("_ParamsT") _YieldT = TypeVar("_YieldT") _T = TypeVar("_T") -_E = TypeVar("_E", bound=BaseException) class _Ok(Generic[_T]): @@ -33,8 +31,8 @@ def __init__(self, value: _T): self.value = value -class _Err(Generic[_E]): - def __init__(self, error: _E): +class _Err(Generic[_T]): + def __init__(self, error: _T): self.error = error From 7f5754306371595444a8ab1b30581e04d05d00a8 Mon Sep 17 00:00:00 2001 From: Callan Gray Date: Wed, 8 Jul 2026 12:08:50 +0800 Subject: [PATCH 3/3] Add iterate exception tests Signed-off-by: Callan Gray --- tests/unit/test_iterate.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/unit/test_iterate.py b/tests/unit/test_iterate.py index 211e687..8e0e755 100644 --- a/tests/unit/test_iterate.py +++ b/tests/unit/test_iterate.py @@ -187,3 +187,37 @@ async def test_iterate_buffer_maxsize(streamcontext, buffer_maxsize: int | None) assert output == TEST_VALUES[:expected_len] await asyncio.wait_for(asyncio.get_running_loop().shutdown_default_executor(), 1.0) + + +class _ReciprocalIterator: + def __init__(self, iterator): + self._iter = iterator + + def __iter__(self): + return self + + def __next__(self): + return 1.0 / next(self._iter) + + +@athreading.iterate +def aiterate_recipricol(iterator): + return _ReciprocalIterator(iterator) + + +@pytest.mark.asyncio +async def test_iterate_background_exception(): + async with aiterate_recipricol(iter(range(-1, 2))) as stream: + with pytest.raises(ZeroDivisionError, match="float division by zero"): + [msg async for msg in stream] + + +@pytest.mark.xfail(reason="not supported yet") +@pytest.mark.asyncio +async def test_iterate_background_exception_suppress(): + async with aiterate_recipricol(iter(range(-1, 2))) as stream: + ait = stream.__aiter__() + assert await ait.__anext__() == -1 + with pytest.raises(ZeroDivisionError, match="float division by zero"): + await ait.__anext__() + assert await ait.__anext__() == 1