feat: sequence method for Stream

pyproject.toml

@@ -55,6 +55,3 @@ papertrail = "papertrail.__main__"
 
 [tool.ty.src]
 include = ["src", "tests"]
-
-[tool.uv]
-exclude-newer = "2 weeks"

src/danom/_new_type.py

@@ -118,10 +118,7 @@ def wrapper(_instance: attrs.AttrsInstance, attribute: attrs.Attribute, value: T
     return wrapper
 
 
-C = TypeVar("C", bound=Callable[P, object])  # type: ignore[invalid-type-form]
-
-
-def _to_list(value: C | Sequence[C] | None) -> list[C]:
+def _to_list[C: Callable[..., object]](value: C | Sequence[C] | None) -> list[C]:
     if value is None:
         return []
 

src/danom/_stream.py

@@ -14,6 +14,8 @@
 
 import attrs
 
+from danom._result import Ok, Result
+
 T = TypeVar("T")
 U = TypeVar("U")
 E = TypeVar("E")
@@ -322,6 +324,24 @@ def partition(
                 neg.append(x)
         return (Stream.from_iterable(pos), Stream.from_iterable(neg))
 
+    def sequence(self, *, workers: int = 1, use_threads: bool = False) -> Result[T, E]:
+        if workers > 1:
+            seq_tuple = self.par_collect(workers=workers, use_threads=use_threads)
+        else:
+            seq_tuple = self.collect()
+
+        if not all(isinstance(res, Result) for res in seq_tuple):
+            raise TypeError("All elements in the `Stream` must be of `Result` type")
+
+        results = []
+
+        for res in seq_tuple:
+            if not res.is_ok():
+                return res
+            results.append(res.unwrap())
+
+        return Ok(tuple(results))
+
     def fold(
         self, initial: T, fn: Callable[[T, U], T], *, workers: int = 1, use_threads: bool = False
     ) -> T:

tests/conftest.py

@@ -5,8 +5,8 @@
 from pathlib import Path
 from typing import Any, NoReturn, Self
 
-from src.danom import safe, safe_method
-from src.danom._result import Err, Ok, Result
+from danom import safe, safe_method
+from danom._result import Err, Ok, Result
 
 REPO_ROOT = Path(__file__).parents[1]
 

tests/test_api.py

@@ -1,6 +1,6 @@
 import pytest
 
-from src.danom import Err, Ok, Result
+from danom import Err, Ok, Result
 
 
 @pytest.mark.parametrize(

tests/test_benchmarks.py

@@ -110,6 +110,12 @@ def _() -> None:
         )
 
 
+def test_stream_of_results_sequence(benchmark) -> None:
+    @benchmark
+    def _() -> None:
+        (Stream.from_iterable([Ok(i) for i in range(100)]).sequence())
+
+
 def test_compose(benchmark) -> None:
     fn = compose(add_one, double, add_one)
 

tests/test_monad_laws.py

@@ -1,7 +1,7 @@
 from hypothesis import given
 from hypothesis import strategies as st
 
-from src.danom import Err, Result
+from danom import Err, Result
 from tests.conftest import safe_add_one, safe_double
 
 inners = st.one_of(

tests/test_new_type.py

@@ -5,8 +5,8 @@
 import pytest
 from hypothesis import given
 
-from src.danom import new_type
-from src.danom._new_type import _validate_bool_func
+from danom import new_type
+from danom._new_type import _validate_bool_func
 from tests.conftest import has_len
 
 

tests/test_result.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from src.danom import Err, Ok, Result
+from danom import Err, Ok, Result
 from tests.conftest import add_one
 
 

tests/test_stream.py

@@ -6,8 +6,9 @@
 from hypothesis import given
 from hypothesis import strategies as st
 
-from src.danom import Stream
-from src.danom._stream import _FILTER, _MAP, _TAP, _apply_fns
+from danom import Stream
+from danom._result import Err, Ok
+from danom._stream import _FILTER, _MAP, _TAP, _apply_fns
 from tests.conftest import (
     REPO_ROOT,
     AsyncValueLogger,
@@ -127,6 +128,40 @@ def test_tap(collect_fn, kwargs):
         assert sorted(values) == [1, 1, 2, 2, 3, 3, 4, 4]
 
 
+@pytest.mark.parametrize(
+    ("kwargs"),
+    [
+        pytest.param({}, id="simple `collect`"),
+        pytest.param({"workers": 4}, id="`par_collect` with workers passed in"),
+        pytest.param({"workers": -1}, id="`par_collect` with n-1 workers"),
+        pytest.param({"use_threads": True}, id="`par_collect` with threads True"),
+    ],
+)
+@pytest.mark.parametrize(
+    ("seq", "expected_result", "expected_context"),
+    [
+        pytest.param(
+            (Ok(0), Ok(1), Ok(2)),
+            Ok((0, 1, 2)),
+            nullcontext(),
+            id="sequence of Oks returns Ok[tuple[T]]",
+        ),
+        pytest.param(
+            (Ok(0), Err(1), Ok(2)), Err(1), nullcontext(), id="returns first Err in the seq"
+        ),
+        pytest.param(
+            (Ok(0), 1, Ok(2)),
+            Ok((0, 1, 2)),
+            pytest.raises(TypeError),
+            id="raises error if not all elements are Result",
+        ),
+    ],
+)
+def test_sequence(kwargs, seq, expected_result, expected_context):
+    with expected_context:
+        assert Stream.from_iterable(seq).sequence(**kwargs) == expected_result
+
+
 # async tests
 
 

tests/test_utils.py

@@ -1,8 +1,8 @@
 import pytest
 from papertrail import example
 
-from src.danom import Ok, compose, identity, invert
-from src.danom._utils import all_of, any_of, none_of
+from danom import Ok, compose, identity, invert
+from danom._utils import all_of, any_of, none_of
 from tests.conftest import add_one, divisible_by_3, divisible_by_5, has_len