great repr

kioss/_pipe.py

@@ -288,7 +288,7 @@ def _accept(self, visitor: "_visitor.AVisitor") -> Any:
         return visitor.visitMapPipe(self)
 
     def __str__(self) -> str:
-        return f"Map(function of type {type(self.func)}, using {self.n_threads} threads)"
+        return f"Map(function of type {type(self.func)}, using {self.n_threads} thread{'s' if self.n_threads > 1 else ''})"
 
 class LogPipe(APipe[T]):
     def __init__(self, upstream: APipe[T], what: str = "elements"):
@@ -310,7 +310,7 @@ def _accept(self, visitor: "_visitor.AVisitor") -> Any:
         return visitor.visitFlattenPipe(self)
 
     def __str__(self) -> str:
-        return f"Flatten(using {self.n_threads} threads)"
+        return f"Flatten(using {self.n_threads} thread{'s' if self.n_threads > 1 else ''})"
 
 class BatchPipe(APipe[List[T]]):
     def __init__(self, upstream: APipe[T], size: int, period: float):
@@ -322,7 +322,7 @@ def _accept(self, visitor: "_visitor.AVisitor") -> Any:
         return visitor.visitBatchPipe(self)
 
     def __str__(self) -> str:
-        return f"Batch(elements by groups of {self.size}, or over a period of {self.period} seconds)"
+        return f"Batch(elements by groups of {self.size} element{'s' if self.size > 1 else ''}, or over a period of {self.period} second{'s' if self.period > 1 else ''})"
 
 class CatchPipe(APipe[T]):
     def __init__(

kioss/_util.py

@@ -1,5 +1,5 @@
 from typing import Any, Callable, Iterable, Iterator, Type, TypeVar, Union
-
+from typing_extensions import TypeGuard
 import logging
 
 LOGGER = logging.getLogger("kioss")
@@ -43,7 +43,7 @@ def identity(obj: T) -> T:
     return obj
 
 
-def duck_check_type_is_iterator(expected_iterator: Any) -> None:
+def duck_check_type_is_iterator(expected_iterator: Any) -> TypeGuard[Iterator]:
     """
     Raises:
         TypeError: If the expected_iterator does not implement __iter__ and __next__ methods.
@@ -72,3 +72,5 @@ def duck_check_type_is_iterator(expected_iterator: Any) -> None:
         raise TypeError(
             f"Provided object is not an iterator because it implements the __iter__ but not the __next__ one."
         )
+
+    return True

kioss/_visitor.py

@@ -98,7 +98,6 @@ def visitLogPipe(self, pipe: _pipe.LogPipe) -> Any:
         return self.visitAnyPipe(pipe)
 
 class IteratorGeneratingVisitor(AVisitor):
-
     def visitSourcePipe(self, pipe: _pipe.SourcePipe[T]) -> Iterator[T]:
         iterator = pipe.source()
         try:
@@ -134,7 +133,7 @@ def visitFilterPipe(self, pipe: _pipe.FilterPipe[T]) -> Iterator[T]:
 
     def visitBatchPipe(self, pipe: _pipe.BatchPipe[T]) -> Iterator[List[T]]:
         return _exec.BatchingIteratorWrapper(
-            iter(pipe.upstream), pipe.size, pipe.period
+            pipe.upstream._accept(self), pipe.size, pipe.period
         )
 
     def visitSlowPipe(self, pipe: _pipe.SlowPipe[T]) -> Iterator[T]: