replace NoopSemaphore() with Semaphore(sys.maxsize)

Author: ebonnal

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,7 +38,7 @@
     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
 
@@ -62,7 +63,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 +77,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:

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:

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

streamable/_tools/_threading.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()