Fixed types for skip_errors.

Author: s3rius

README.md

@@ -200,8 +200,15 @@ def to_string(val: Any) -> str:
 
 async def main():
     pipe = (
-        Pipeline(broker).group(
-            Group()
+        Pipeline(broker)
+        .group(
+            Group(
+                # 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 1
             .add(add_one, 1)
             # Here's a task that multiplies 2 by 2

taskiq_pipelines/pipeliner.py

@@ -56,7 +56,7 @@ class Pipeline(Generic[_FuncParams, _ReturnType]):
     def __init__(
         self: "Pipeline[[], _ReturnType]",
         broker: AsyncBroker,
-        task: Optional[Group[_ReturnType]] = None,
+        task: Optional[Group[Any, _ReturnType]] = None,
     ) -> None: ...
 
     @overload
@@ -78,7 +78,7 @@ def __init__(
             Union[
                 AsyncKicker[_FuncParams, _ReturnType],
                 AsyncTaskiqDecoratedTask[_FuncParams, _ReturnType],
-                Group[_ReturnType],
+                Group[Any, _ReturnType],
             ]
         ] = None,
     ) -> None:
@@ -353,7 +353,7 @@ def filter(
 
     def group(
         self: "Pipeline[_FuncParams, _ReturnType]",
-        group: Group[_T2],
+        group: Group[Any, _T2],
     ) -> "Pipeline[_FuncParams, _T2]":
         """
         Add group task execution step.

taskiq_pipelines/task_group.py

@@ -1,5 +1,5 @@
 from types import CoroutineType
-from typing import Any, Coroutine, Generic, Tuple, Union, overload
+from typing import Any, Coroutine, Generic, Literal, Optional, Tuple, Union, overload
 
 from taskiq import AsyncTaskiqDecoratedTask
 from taskiq.kicker import AsyncKicker
@@ -9,11 +9,13 @@
 
 _Tups = TypeVarTuple("_Tups")
 _T = TypeVar("_T")
+# Whether skip_errors is set to True or False
+_S = TypeVar("_S", bound=bool)
 _TVal = TypeVar("_TVal")
 _Params = ParamSpec("_Params")
 
 
-class Group(Generic[_T]):
+class Group(Generic[_S, _T]):
     """
     Group of tasks.
 
@@ -23,8 +25,22 @@ class Group(Generic[_T]):
     :param skip_errors: If True, errors in one task will not affect others.
     """
 
+    @overload
+    def __init__(
+        self: "Group[Literal[True], Tuple[()]]",
+        skip_errors: Literal[True],
+        check_interval: float = 0.1,
+    ) -> None: ...
+
+    @overload
     def __init__(
-        self: "Group[Tuple[()]]",
+        self: "Group[Literal[False], Tuple[()]]",
+        skip_errors: bool = False,
+        check_interval: float = 0.1,
+    ) -> None: ...
+
+    def __init__(
+        self: "Group[Any, Tuple[()]]",
         skip_errors: bool = False,
         check_interval: float = 0.1,
     ) -> None:
@@ -34,7 +50,7 @@ def __init__(
 
     @overload
     def add(
-        self: "Group[Tuple[Unpack[_Tups]]]",
+        self: "Group[Literal[True], Tuple[Unpack[_Tups]]]",
         task: Union[
             AsyncKicker[_Params, Coroutine[Any, Any, _TVal]],
             AsyncKicker[_Params, "CoroutineType[Any, Any, _TVal]"],
@@ -43,21 +59,45 @@ def add(
         ],
         *args: _Params.args,
         **kwargs: _Params.kwargs,
-    ) -> "Group[Tuple[Unpack[_Tups], _TVal]]": ...
+    ) -> "Group[_S, Tuple[Unpack[_Tups], Optional[_TVal]]]": ...
+
+    @overload
+    def add(
+        self: "Group[Literal[False], Tuple[Unpack[_Tups]]]",
+        task: Union[
+            AsyncKicker[_Params, Coroutine[Any, Any, _TVal]],
+            AsyncKicker[_Params, "CoroutineType[Any, Any, _TVal]"],
+            AsyncTaskiqDecoratedTask[_Params, Coroutine[Any, Any, _TVal]],
+            AsyncTaskiqDecoratedTask[_Params, "CoroutineType[Any, Any, _TVal]"],
+        ],
+        *args: _Params.args,
+        **kwargs: _Params.kwargs,
+    ) -> "Group[_S, Tuple[Unpack[_Tups], _TVal]]": ...
+
+    @overload
+    def add(
+        self: "Group[Literal[True], Tuple[Unpack[_Tups]]]",
+        task: Union[
+            AsyncKicker[_Params, _TVal],
+            AsyncTaskiqDecoratedTask[_Params, _TVal],
+        ],
+        *args: _Params.args,
+        **kwargs: _Params.kwargs,
+    ) -> "Group[_S, Tuple[Unpack[_Tups], Optional[_TVal]]]": ...
 
     @overload
     def add(
-        self: "Group[Tuple[Unpack[_Tups]]]",
+        self: "Group[Literal[False], Tuple[Unpack[_Tups]]]",
         task: Union[
             AsyncKicker[_Params, _TVal],
             AsyncTaskiqDecoratedTask[_Params, _TVal],
         ],
         *args: _Params.args,
         **kwargs: _Params.kwargs,
-    ) -> "Group[Tuple[Unpack[_Tups], _TVal]]": ...
+    ) -> "Group[_S, Tuple[Unpack[_Tups], _TVal]]": ...
 
     def add(
-        self: "Group[Any]",
+        self: "Group[Any, Any]",
         task: Union[AsyncKicker[_Params, Any], AsyncTaskiqDecoratedTask[_Params, Any]],
         *args: _Params.args,
         **kwargs: _Params.kwargs,