diff --git a/src/danom/_result.py b/src/danom/_result.py index f8b431d..69d8d63 100644 --- a/src/danom/_result.py +++ b/src/danom/_result.py @@ -264,7 +264,9 @@ def map[**P](self, func: Mappable, *args: P.args, **kwargs: P.kwargs) -> Self: return self def map_err[**P](self, func: Mappable, *args: P.args, **kwargs: P.kwargs) -> Err[E_co]: - return Err(func(self.error, *args, **kwargs)) + return Err( + func(self.error, *args, **kwargs), input_args=self.input_args, traceback=self.traceback + ) def and_then[**P](self, func: Bindable, *args: P.args, **kwargs: P.kwargs) -> Self: # noqa: ARG002 return self diff --git a/src/danom/_stream.py b/src/danom/_stream.py index 66e3898..92ca9e6 100644 --- a/src/danom/_stream.py +++ b/src/danom/_stream.py @@ -89,7 +89,7 @@ def __bool__(self) -> bool: @attrs.define(frozen=True) class Stream[T](_BaseStream): - """An immutable lazy iterator with functional operations. + """A lazy iterator with functional operations. Why bother? ----------- @@ -472,6 +472,8 @@ def par_collect(self, workers: int = 4, *, use_threads: bool = False) -> tuple[U if workers == -1: workers = (os.cpu_count() or 5) - 1 + workers = max(workers, 1) + executor_cls = ThreadPoolExecutor if use_threads else ProcessPoolExecutor batches = [ diff --git a/tests/test_stream.py b/tests/test_stream.py index 6a56247..d531f02 100644 --- a/tests/test_stream.py +++ b/tests/test_stream.py @@ -34,6 +34,9 @@ def _get_attr_collect(stream: Stream, collect_fn: str, kwargs: dict) -> tuple: pytest.param("collect", {}, id="simple `collect`"), pytest.param("par_collect", {"workers": 4}, id="`par_collect` with workers passed in"), pytest.param("par_collect", {"workers": -1}, id="`par_collect` with n-1 workers"), + pytest.param( + "par_collect", {"workers": 0}, id="`par_collect` with 0 workers falls back to 1 worker" + ), pytest.param("par_collect", {"use_threads": True}, id="`par_collect` with threads True"), ], )