Add empty param name support

Author: cofob

taskiq_pipelines/__init__.py

@@ -1,4 +1,5 @@
 """Pipelines for taskiq tasks."""
+from taskiq_pipelines.constants import EMPTY_PARAM_NAME
 from taskiq_pipelines.exceptions import AbortPipeline, PipelineError
 from taskiq_pipelines.middleware import PipelineMiddleware
 from taskiq_pipelines.pipeliner import Pipeline
@@ -8,4 +9,5 @@
     "PipelineError",
     "AbortPipeline",
     "PipelineMiddleware",
+    "EMPTY_PARAM_NAME",
 ]

taskiq_pipelines/constants.py

@@ -1,2 +1,6 @@
+from typing import Literal
+
 CURRENT_STEP = "_pipe_current_step"
 PIPELINE_DATA = "_pipe_data"
+
+EMPTY_PARAM_NAME: Literal[-1] = -1

taskiq_pipelines/pipeliner.py

@@ -1,5 +1,15 @@
 import json
-from typing import Any, Coroutine, Generic, List, Optional, TypeVar, Union, overload
+from typing import (
+    Any,
+    Coroutine,
+    Generic,
+    List,
+    Literal,
+    Optional,
+    TypeVar,
+    Union,
+    overload,
+)
 
 import pydantic
 from taskiq import AsyncBroker, AsyncTaskiqTask
@@ -58,7 +68,7 @@ def call_next(
             AsyncKicker[Any, Coroutine[Any, Any, _T2]],
             AsyncTaskiqDecoratedTask[Any, Coroutine[Any, Any, _T2]],
         ],
-        param_name: Optional[str] = None,
+        param_name: Union[Optional[str], Literal[-1]] = None,
         **additional_kwargs: Any,
     ) -> "Pipeline[_FuncParams, _T2]":
         ...
@@ -70,7 +80,7 @@ def call_next(
             AsyncKicker[Any, _T2],
             AsyncTaskiqDecoratedTask[Any, _T2],
         ],
-        param_name: Optional[str] = None,
+        param_name: Union[Optional[str], Literal[-1]] = None,
         **additional_kwargs: Any,
     ) -> "Pipeline[_FuncParams, _T2]":
         ...
@@ -81,7 +91,7 @@ def call_next(
             AsyncKicker[Any, Any],
             AsyncTaskiqDecoratedTask[Any, Any],
         ],
-        param_name: Optional[str] = None,
+        param_name: Union[Optional[str], Literal[-1]] = None,
         **additional_kwargs: Any,
     ) -> Any:
         """
@@ -94,7 +104,8 @@ def call_next(
         if param_name is specified.
 
         :param task: task to execute.
-        :param param_name: kwarg param name, defaults to None
+        :param param_name: kwarg param name, defaults to None.
+            If set to -1 (EMPTY_PARAM_NAME), result is not passed.
         :param additional_kwargs: additional kwargs to task.
         :return: updated pipeline.
         """

taskiq_pipelines/steps/sequential.py

@@ -1,11 +1,11 @@
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Literal, Optional, Union
 
 import pydantic
 from taskiq import AsyncBroker, AsyncTaskiqDecoratedTask, TaskiqResult
 from taskiq.kicker import AsyncKicker
 
 from taskiq_pipelines.abc import AbstractStep
-from taskiq_pipelines.constants import CURRENT_STEP, PIPELINE_DATA
+from taskiq_pipelines.constants import CURRENT_STEP, EMPTY_PARAM_NAME, PIPELINE_DATA
 
 
 class SequentialStep(pydantic.BaseModel, AbstractStep, step_name="sequential"):
@@ -19,9 +19,17 @@ class SequentialStep(pydantic.BaseModel, AbstractStep, step_name="sequential"):
 
     task_name: str
     labels: Dict[str, str]
-    param_name: Optional[str]
+    # order is important here, otherwise pydantic will always choose str.
+    # we use int instead of Literal[-1] because pydantic thinks that -1 is always str.
+    param_name: Union[Optional[int], str]
     additional_kwargs: Dict[str, Any]
 
+    @pydantic.validator("param_name")
+    def validate_param_name(cls, value: Union[Optional[str], int]) -> Union[Optional[str], int]:
+        if isinstance(value, int) and value != EMPTY_PARAM_NAME:
+            raise ValueError("must be str, None or -1 (EMPTY_PARAM_NAME)")
+        return value
+
     def dumps(self) -> str:
         """
         Dumps step as string.
@@ -78,9 +86,11 @@ async def act(
                 **{PIPELINE_DATA: pipe_data, CURRENT_STEP: step_number},  # type: ignore
             )
         )
-        if self.param_name:
+        if isinstance(self.param_name, str):
             self.additional_kwargs[self.param_name] = result.return_value
             await kicker.kiq(**self.additional_kwargs)
+        elif self.param_name == EMPTY_PARAM_NAME:
+            await kicker.kiq(**self.additional_kwargs)
         else:
             await kicker.kiq(result.return_value, **self.additional_kwargs)
 
@@ -91,7 +101,7 @@ def from_task(
             AsyncKicker[Any, Any],
             AsyncTaskiqDecoratedTask[Any, Any],
         ],
-        param_name: Optional[str],
+        param_name: Union[Optional[str], int],
         **additional_kwargs: Any,
     ) -> "SequentialStep":
         """