Just a concept of gather-like method

CHANGELOG.md

@@ -14,9 +14,10 @@ See [0Ver](https://0ver.org/).
 - Improve inference of `ResultLike` objects when exception catching
   decorator is applied with explicit exception types
 - Add picky exceptions to `impure_safe` decorator like `safe` has. Issue #1543
-- Add partition function to result module. Issue #1905
+- Add `partition` function to methods module. Issue #1905
 - Adds `default_error` parameter to `returns.converters.maybe_to_result`,
   which provides a default error value for `Failure`
+- Add `returns.methods.gather` utility method
 
 
 ### Misc

docs/pages/methods.rst

@@ -79,19 +79,44 @@ Here's a full example:
 partition
 ~~~~~~~~~
 
-:func:`partition <returns.result.partition>` is used to convert
+:func:`partition <returns.methods.partition>` is used to convert
 list of :class:`~returns.interfaces.Unwrappable`
 instances like :class:`~returns.result.Result`,
 :class:`~returns.io.IOResult`, and :class:`~returns.maybe.Maybe`
 to a tuple of two lists: successes and failures.
 
 .. code:: python
 
-    >>> from returns.result import Failure, Success
-    >>> from returns.methods import partition
-    >>> results = [Success(1), Failure(2), Success(3), Failure(4)]
-    >>> partition(results)
-    ([1, 3], [2, 4])
+  >>> from returns.result import Failure, Success
+  >>> from returns.methods import partition
+  >>> results = [Success(1), Failure(2), Success(3), Failure(4)]
+  >>> partition(results)
+  ([1, 3], [2, 4])
+
+gather
+~~~~~~
+
+:func:`gather <returns.methods.gather>` is used to safely concurrently
+execute multiple awaitable objects(any object with ``__await__`` method,
+included function marked with async keyword) and return a tuple of wrapped results
+:class: `~returns.io.IOResult`.
+Embrace railway-oriented programming princple of executing as many IO operations
+as possible before synchrounous computations.
+
+.. code:: python
+
+  >>> import anyio
+  >>> from returns.io import IO, IOSuccess, IOFailure
+  >>> from returns.methods import gather
+
+  >>> async def coro():
+  ...    return 1
+  >>> async def coro_raise():
+  ...    raise ValueError(2)
+  >>> anyio.run(gather,[coro(), coro_raise()])
+  (IOSuccess(1), IOFailure(ValueError(2)))
+
+
 
 API Reference
 -------------

returns/methods/__init__.py

@@ -1,4 +1,5 @@
 from returns.methods.cond import cond as cond
+from returns.methods.gather import gather as gather
 from returns.methods.partition import partition as partition
 from returns.methods.unwrap_or_failure import (
     unwrap_or_failure as unwrap_or_failure,

returns/methods/gather.py

@@ -1,11 +1,10 @@
-from typing import Any
 
 from typing import Awaitable, Iterable
 
 import anyio
-from returns.future import Future, FutureResult
-from returns.io import IOResult
 
+from returns.future import FutureResult
+from returns.io import IOResult
 
 
 async def gather(
@@ -24,20 +23,24 @@ async def gather(
 
       >>> async def coro():
       ...    return 1
-      >>> assert anyio.run(gather([coro()])) == (IOSuccess(1), )
+      >>> assert anyio.run(gather, [coro()]) == (IOSuccess(1), )
       >>> container = FutureResult(coro())
-      >>> assert anyio.run(gather([container.awaitable])) == (IOSuccess(1), )
+      >>> assert anyio.run(gather, [container.awaitable]) == (IOSuccess(1), )
 
     """
-
     async with anyio.create_task_group() as tg:
         containers_t = tuple(containers)
-        results: list[IOResult] = len(containers_t)*[IOResult(None)]
+        ioresults: dict[int, IOResult] = {}
 
-        async def run_task(coro: Awaitable, index: int):
-            results[index] = await FutureResult(coro)
+        async def _coro_wrapper(coro: Awaitable):  # noqa: WPS430
+            try:
+                return IOResult.from_value(await coro)
+            except Exception as exc:
+                return IOResult.from_failure(exc)
 
-        for i, coro in enumerate(containers_t):
-            tg.start_soon(run_task, coro, i)
-    return tuple(results)
+        async def _run_task(coro: Awaitable, index: int):  # noqa: WPS430
+            ioresults[index] = await _coro_wrapper(coro)
 
+        for coro_index, coro in enumerate(containers_t):
+            tg.start_soon(_run_task, coro, coro_index)
+    return tuple([ioresults[key] for key in sorted(ioresults.keys())])

tests/test_methods/test_gather.py

@@ -0,0 +1,22 @@
+
+import pytest
+
+from returns.future import Future, FutureResult
+from returns.io import IO, IOResult
+from returns.methods import gather
+
+
+@pytest.mark.parametrize(('containers', 'expected'), [
+    (
+        (
+            Future.from_value(1),
+            FutureResult.from_value(2),
+            FutureResult.from_failure(None),
+        ),
+        (IO(1), IOResult.from_value(2), FutureResult.from_failure(None)),
+    ),
+    ((), ()),
+])
+def test_gather(containers, expected):
+    """Test partition function."""
+    assert gather(containers) == expected