Fixed a bug, added docs.

Author: s3rius

README.md

@@ -227,3 +227,66 @@ if __name__ == "__main__":
     asyncio.run(main())
 
 ```
+
+Aletrnatively, you can use `GroupWithArgs` class to pass arguments of previous step
+as an argument to called functions. Here's an example:
+
+```python
+import asyncio
+from typing import Any
+
+from taskiq.brokers.inmemory_broker import InMemoryBroker
+
+from taskiq_pipelines import GroupWithArgs, Pipeline, PipelineMiddleware
+
+broker = InMemoryBroker()
+broker.add_middlewares(PipelineMiddleware())
+
+
+@broker.task
+def add_one(value: int) -> int:
+    return value + 1
+
+
+@broker.task
+def mul_two(val: int) -> int:
+    return val * 2
+
+
+@broker.task
+def to_string(val: Any) -> str:
+    return str(val)
+
+
+async def main():
+    pipe = (
+        # The pipelines starts with `add_one` task.
+        Pipeline(broker, add_one)
+        # All values are passed to the group
+        .group(
+            GroupWithArgs(
+                # Aborts pipeline
+                # if any of tasks fails
+                skip_errors=False,
+                # How often to check for completion.
+                check_interval=0.1,
+            )
+            # Here we start task that adds 1 to result
+            # of the previous task.
+            # The result is passed as keyword argument "value"
+            .add(add_one, param_name="value")
+            # Here's a task that multiplies 2
+            .add(mul_two)
+        # Here we map all results to string
+        ).map(to_string)
+    )
+    task = await pipe.kiq(1)
+    result = await task.wait_result()
+    # Here it should output
+    # Calculated value: ['3', '4']
+    print("Calculated value:", result.return_value)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```

taskiq_pipelines/__init__.py

@@ -3,11 +3,12 @@
 from taskiq_pipelines.exceptions import AbortPipeline, PipelineError
 from taskiq_pipelines.middleware import PipelineMiddleware
 from taskiq_pipelines.pipeliner import Pipeline
-from taskiq_pipelines.task_group import Group
+from taskiq_pipelines.task_group import Group, GroupWithArgs
 
 __all__ = [
     "AbortPipeline",
     "Group",
+    "GroupWithArgs",
     "Pipeline",
     "PipelineError",
     "PipelineMiddleware",

taskiq_pipelines/steps/group.py

@@ -20,7 +20,7 @@
 async def wait_group_tasks(
     task_ids: List[str],
     check_interval: float,
-    skip_errors: bool = True,
+    skip_errors: bool = False,
     context: Context = TaskiqDepends(),
 ) -> tuple[Any, ...]:
     """Waits for subtasks to complete."""
@@ -130,7 +130,7 @@ async def act(
             )
             .kiq(
                 task_ids=ids,
-                skip_errors=True,
+                skip_errors=self.skip_errors,
                 check_interval=self.check_interval,
             )
         )