From 08d9f196e313af630a26fe952705fcc3d9e37107 Mon Sep 17 00:00:00 2001 From: ebonnal Date: Sat, 2 May 2026 13:35:17 +0100 Subject: [PATCH 1/2] `.buffer`: make `up_to` optional --- streamable/_afunctions.py | 2 +- streamable/_aiterators.py | 10 ++++++---- streamable/_functions.py | 2 +- streamable/_iterators.py | 7 ++++--- streamable/_stream.py | 13 ++++++------- streamable/_tools/_async.py | 18 ++++++++++++++++++ streamable/_tools/_threading.py | 17 +++++++++++++++++ tests/test_buffer.py | 11 ++++++++--- tests/test_tools.py | 16 ++++++++++++++++ 9 files changed, 77 insertions(+), 19 deletions(-) create mode 100644 streamable/_tools/_threading.py diff --git a/streamable/_afunctions.py b/streamable/_afunctions.py index 5ff792c7..b581838b 100644 --- a/streamable/_afunctions.py +++ b/streamable/_afunctions.py @@ -33,7 +33,7 @@ def buffer( aiterator: AsyncIterator[T], - up_to: int, + up_to: Optional[int] = None, ) -> AsyncIterator[T]: return _aiterators.BufferAsyncIterator(aiterator, up_to) diff --git a/streamable/_aiterators.py b/streamable/_aiterators.py index eee57894..0d75755a 100644 --- a/streamable/_aiterators.py +++ b/streamable/_aiterators.py @@ -37,7 +37,7 @@ FIFOFutureResults, FutureResults, ) -from streamable._tools._async import AsyncFunction, empty_aiter +from streamable._tools._async import AsyncFunction, NoopSemaphore, empty_aiter from streamable._tools._context import noop_context_manager from streamable._tools._error import ExceptionContainer, RaisingAsyncIterator @@ -59,7 +59,7 @@ class _BufferAsyncIterable(AsyncIterable[Union[T, ExceptionContainer]]): def __init__( self, iterator: AsyncIterator[T], - up_to: int, + up_to: Optional[int], ) -> None: self.iterator = iterator self.up_to = up_to @@ -76,7 +76,9 @@ def _lazy_buffer(self) -> "asyncio.Queue[Union[T, ExceptionContainer]]": @property def _lazy_slots(self) -> asyncio.Semaphore: if not self._slots: - self._slots = asyncio.Semaphore(self.up_to) + self._slots = ( + asyncio.Semaphore(self.up_to) if self.up_to else NoopSemaphore() + ) return self._slots async def _buffer_upstream(self) -> None: @@ -115,7 +117,7 @@ class BufferAsyncIterator(RaisingAsyncIterator[T]): def __init__( self, iterator: AsyncIterator[T], - up_to: int, + up_to: Optional[int], ) -> None: super().__init__(_BufferAsyncIterable(iterator, up_to).__aiter__()) diff --git a/streamable/_functions.py b/streamable/_functions.py index 6dd7cc14..af1e26b2 100644 --- a/streamable/_functions.py +++ b/streamable/_functions.py @@ -30,7 +30,7 @@ def buffer( iterator: Iterator[T], - up_to: int, + up_to: Optional[int] = None, ) -> Iterator[T]: return _iterators.BufferIterator(iterator, up_to) diff --git a/streamable/_iterators.py b/streamable/_iterators.py index 8f6d8936..742564dc 100644 --- a/streamable/_iterators.py +++ b/streamable/_iterators.py @@ -27,6 +27,7 @@ from streamable._tools._context import noop_context_manager from streamable._tools._observation import Observation +from streamable._tools._threading import NoopSemaphore from streamable._tools._sentinel import STOP_ITERATION from streamable._tools._validation import validate_sync_flatten_iterable @@ -55,11 +56,11 @@ class _BufferIterable(Iterable[Union[T, ExceptionContainer]]): def __init__( self, iterator: Iterator[T], - up_to: int, + up_to: Optional[int], ) -> None: self.iterator = iterator self._buffer: "queue.Queue[Union[T, ExceptionContainer]]" = queue.Queue() - self._slots = Semaphore(up_to) + self._slots = Semaphore(up_to) if up_to else NoopSemaphore() self._stopped = False def _buffer_upstream(self) -> None: @@ -99,7 +100,7 @@ class BufferIterator(RaisingIterator[T]): def __init__( self, iterator: Iterator[T], - up_to: int, + up_to: Optional[int], ) -> None: super().__init__(_BufferIterable(iterator, up_to).__iter__()) diff --git a/streamable/_stream.py b/streamable/_stream.py index 3a17b945..504a1763 100644 --- a/streamable/_stream.py +++ b/streamable/_stream.py @@ -300,17 +300,15 @@ def cast(self, into: Type[U]) -> "stream[U]": def buffer( self, - up_to: int, + up_to: Optional[int] = None, ) -> "stream[T]": """ - Buffer upstream elements into a bounded queue (max size ``up_to``), via a background task. - - Allow to decouple the upstream production rate from the downstream consumption rate. + Buffer upstream elements into a queue, via a background task, decoupling upstream production rate from downstream consumption rate. The background task is a thread during a sync iteration, and an async task during an async iteration. Args: - up_to (``int``): The buffer size. Must be >= 1. When reached, upstream pulling pauses until an element is yielded out of the buffer. + up_to (``int | None``): The buffer size, must be >= 1 when set. When reached, upstream pulling pauses until an element is yielded out of the buffer. Returns: ``stream[T]``: Upstream with buffering. @@ -327,7 +325,8 @@ def buffer( time.sleep(1e-3) assert pulled == [0, 1, 2, 3, 4, 5] """ - validate_int(up_to, gte=1, name="up_to") + if up_to is not None: + validate_int(up_to, gte=1, name="up_to") return BufferStream(self, up_to) @overload @@ -1082,7 +1081,7 @@ class BufferStream(DownStream[T, T]): def __init__( self, upstream: stream[T], - up_to: int, + up_to: Optional[int], ) -> None: super().__init__(upstream) self._up_to = up_to diff --git a/streamable/_tools/_async.py b/streamable/_tools/_async.py index d11e456d..f7c2606b 100644 --- a/streamable/_tools/_async.py +++ b/streamable/_tools/_async.py @@ -1,9 +1,11 @@ +from asyncio import Semaphore from typing import ( Any, AsyncIterator, Awaitable, Callable, Coroutine, + Literal, TypeVar, ) @@ -25,3 +27,19 @@ async def awaitable_to_coroutine(aw: Awaitable[T]) -> T: async def empty_aiter() -> AsyncIterator[Any]: return yield # pragma: no cover + + +class NoopSemaphore(Semaphore): + __slots__ = () + + def __init__(self) -> None: + pass + + async def acquire(self) -> Literal[True]: + return True + + def release(self) -> None: + return + + def locked(self) -> bool: + return False diff --git a/streamable/_tools/_threading.py b/streamable/_tools/_threading.py new file mode 100644 index 00000000..9268a845 --- /dev/null +++ b/streamable/_tools/_threading.py @@ -0,0 +1,17 @@ +from threading import Semaphore + + +class NoopSemaphore(Semaphore): + __slots__ = () + + def __init__(self) -> None: + pass + + def acquire(self, blocking=True, timeout=None) -> bool: + return True + + def release(self, n=1) -> None: + return + + def locked(self) -> bool: + return False diff --git a/tests/test_buffer.py b/tests/test_buffer.py index 620d4da6..21808697 100644 --- a/tests/test_buffer.py +++ b/tests/test_buffer.py @@ -1,3 +1,4 @@ +import sys from typing import Any, AsyncIterable, Callable, Iterable, List import pytest @@ -26,7 +27,7 @@ def test_buffer_preserves_elements(itype: IterableType) -> None: assert alist_or_list(ints.buffer(5), itype) == list(INTEGERS) -@pytest.mark.parametrize("buffer_size", [1, 10]) +@pytest.mark.parametrize("buffer_size", [1, 10, None]) @pytest.mark.parametrize( "itype, slow_identity", [(Iterable, slow_identity), (AsyncIterable, async_slow_identity)], @@ -39,9 +40,13 @@ def test_buffer_size_is_respected( buffering_ints_iter = aiter_or_iter(buffering_ints, itype) assert buffered == [] assert anext_or_next(buffering_ints_iter, itype) == 0 - assert buffered == list(INTEGERS)[: buffer_size + 1] + assert ( + buffered == list(INTEGERS)[: (buffer_size + 1) if buffer_size else sys.maxsize] + ) assert anext_or_next(buffering_ints_iter, itype) == 1 - assert buffered == list(INTEGERS)[: buffer_size + 2] + assert ( + buffered == list(INTEGERS)[: (buffer_size + 2) if buffer_size else sys.maxsize] + ) @pytest.mark.parametrize( diff --git a/tests/test_tools.py b/tests/test_tools.py index 96097283..bfbdf7e7 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -2,6 +2,7 @@ import pytest from streamable._tools._func import sidify from streamable._tools._logging import logfmt_str_escape +from streamable._tools import _async, _threading from typing import Any, Callable, List @@ -88,3 +89,18 @@ def test_logfmt_str_escape(): assert logfmt_str_escape("in ts") == '"in ts"' assert logfmt_str_escape("in\\ts") == r'"in\\ts"' assert logfmt_str_escape('"ints"') == r'"\"ints\""' + + +def test_noop_semaphore() -> None: + semaphore = _threading.NoopSemaphore() + assert semaphore.acquire() + semaphore.release() + assert not semaphore.locked() + + +@pytest.mark.asyncio +async def test_noop_semaphore_async() -> None: + semaphore = _async.NoopSemaphore() + assert await semaphore.acquire() + semaphore.release() + assert not semaphore.locked() From e5bb4a427670fa231e7473efc8e55cbc4a4e7ae3 Mon Sep 17 00:00:00 2001 From: ebonnal Date: Sat, 2 May 2026 16:31:07 +0100 Subject: [PATCH 2/2] replace `NoopSemaphore()` with `Semaphore(sys.maxsize)` --- streamable/_aiterators.py | 10 ++++------ streamable/_iterators.py | 5 +++-- streamable/_stream.py | 2 +- streamable/_tools/_async.py | 27 +-------------------------- streamable/_tools/_threading.py | 17 ----------------- tests/test_tools.py | 16 ---------------- 6 files changed, 9 insertions(+), 68 deletions(-) delete mode 100644 streamable/_tools/_threading.py diff --git a/streamable/_aiterators.py b/streamable/_aiterators.py index 0d75755a..6a90ead0 100644 --- a/streamable/_aiterators.py +++ b/streamable/_aiterators.py @@ -2,6 +2,7 @@ from asyncio.futures import Future from contextlib import suppress import datetime +import sys import time from abc import ABC, abstractmethod from collections import defaultdict, deque @@ -37,11 +38,10 @@ FIFOFutureResults, FutureResults, ) -from streamable._tools._async import AsyncFunction, NoopSemaphore, empty_aiter +from streamable._tools._async import AsyncFunction, anext, empty_aiter from streamable._tools._context import noop_context_manager from streamable._tools._error import ExceptionContainer, RaisingAsyncIterator -from streamable._tools._async import anext T = TypeVar("T") U = TypeVar("U") @@ -62,7 +62,7 @@ def __init__( up_to: Optional[int], ) -> None: self.iterator = iterator - self.up_to = up_to + self.up_to = up_to or sys.maxsize self._buffer: "Optional[asyncio.Queue[Union[T, ExceptionContainer]]]" = None self._slots: Optional[asyncio.Semaphore] = None self._stopped = False @@ -76,9 +76,7 @@ def _lazy_buffer(self) -> "asyncio.Queue[Union[T, ExceptionContainer]]": @property def _lazy_slots(self) -> asyncio.Semaphore: if not self._slots: - self._slots = ( - asyncio.Semaphore(self.up_to) if self.up_to else NoopSemaphore() - ) + self._slots = asyncio.Semaphore(self.up_to) return self._slots async def _buffer_upstream(self) -> None: diff --git a/streamable/_iterators.py b/streamable/_iterators.py index 742564dc..38778339 100644 --- a/streamable/_iterators.py +++ b/streamable/_iterators.py @@ -1,5 +1,6 @@ import datetime import queue +import sys from threading import Semaphore, Thread import time from abc import ABC, abstractmethod @@ -27,7 +28,6 @@ from streamable._tools._context import noop_context_manager from streamable._tools._observation import Observation -from streamable._tools._threading import NoopSemaphore from streamable._tools._sentinel import STOP_ITERATION from streamable._tools._validation import validate_sync_flatten_iterable @@ -59,8 +59,9 @@ def __init__( up_to: Optional[int], ) -> None: self.iterator = iterator + up_to = up_to or sys.maxsize self._buffer: "queue.Queue[Union[T, ExceptionContainer]]" = queue.Queue() - self._slots = Semaphore(up_to) if up_to else NoopSemaphore() + self._slots = Semaphore(up_to) self._stopped = False def _buffer_upstream(self) -> None: diff --git a/streamable/_stream.py b/streamable/_stream.py index 504a1763..e4faf752 100644 --- a/streamable/_stream.py +++ b/streamable/_stream.py @@ -308,7 +308,7 @@ def buffer( The background task is a thread during a sync iteration, and an async task during an async iteration. Args: - up_to (``int | None``): The buffer size, must be >= 1 when set. When reached, upstream pulling pauses until an element is yielded out of the buffer. + up_to (``int | None``): The buffer size. Must be >= 1. When reached, upstream pulling pauses until an element is yielded out of the buffer. Returns: ``stream[T]``: Upstream with buffering. diff --git a/streamable/_tools/_async.py b/streamable/_tools/_async.py index f7c2606b..931406c3 100644 --- a/streamable/_tools/_async.py +++ b/streamable/_tools/_async.py @@ -1,13 +1,4 @@ -from asyncio import Semaphore -from typing import ( - Any, - AsyncIterator, - Awaitable, - Callable, - Coroutine, - Literal, - TypeVar, -) +from typing import Any, AsyncIterator, Awaitable, Callable, Coroutine, TypeVar T = TypeVar("T") R = TypeVar("R") @@ -27,19 +18,3 @@ async def awaitable_to_coroutine(aw: Awaitable[T]) -> T: async def empty_aiter() -> AsyncIterator[Any]: return yield # pragma: no cover - - -class NoopSemaphore(Semaphore): - __slots__ = () - - def __init__(self) -> None: - pass - - async def acquire(self) -> Literal[True]: - return True - - def release(self) -> None: - return - - def locked(self) -> bool: - return False diff --git a/streamable/_tools/_threading.py b/streamable/_tools/_threading.py deleted file mode 100644 index 9268a845..00000000 --- a/streamable/_tools/_threading.py +++ /dev/null @@ -1,17 +0,0 @@ -from threading import Semaphore - - -class NoopSemaphore(Semaphore): - __slots__ = () - - def __init__(self) -> None: - pass - - def acquire(self, blocking=True, timeout=None) -> bool: - return True - - def release(self, n=1) -> None: - return - - def locked(self) -> bool: - return False diff --git a/tests/test_tools.py b/tests/test_tools.py index bfbdf7e7..96097283 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -2,7 +2,6 @@ import pytest from streamable._tools._func import sidify from streamable._tools._logging import logfmt_str_escape -from streamable._tools import _async, _threading from typing import Any, Callable, List @@ -89,18 +88,3 @@ def test_logfmt_str_escape(): assert logfmt_str_escape("in ts") == '"in ts"' assert logfmt_str_escape("in\\ts") == r'"in\\ts"' assert logfmt_str_escape('"ints"') == r'"\"ints\""' - - -def test_noop_semaphore() -> None: - semaphore = _threading.NoopSemaphore() - assert semaphore.acquire() - semaphore.release() - assert not semaphore.locked() - - -@pytest.mark.asyncio -async def test_noop_semaphore_async() -> None: - semaphore = _async.NoopSemaphore() - assert await semaphore.acquire() - semaphore.release() - assert not semaphore.locked()