Merge branch 'release/0.1.1'

Author: s3rius

.flake8

@@ -88,6 +88,8 @@ ignore =
   ; Consider possible security implications associated with pickle module
   ; Pickle and modules that wrap it can be unsafe when used to deserialize untrusted data, possible security issue
   S403, S301
+  ; Found too many imported names from a module
+  WPS235
 
 per-file-ignores =
   ; all tests
@@ -104,6 +106,10 @@ per-file-ignores =
   DAR101,
   ; Found too many arguments
   WPS211,
+  ; Found nested function
+  WPS430,
+  ; Found too short name
+  WPS111,
 
   ; all init files
   __init__.py:

.github/workflows/test.yml

@@ -1,6 +1,6 @@
 name: Testing taskiq-pipelines
 
-on: push
+on: pull_request
 
 jobs:
   black:

.pre-commit-config.yaml

@@ -41,7 +41,7 @@ repos:
         language: system
         pass_filenames: false
         types: [python]
-        args: [--count, taskiq_pipelines]
+        args: [--count, taskiq_pipelines, tests]
 
       - id: mypy
         name: Validate types with MyPy

README.md

@@ -138,11 +138,14 @@ We have a few steps available for chaining calls:
 This type of step is just an ordinary call of the function.
 If you haven't specified `param_name` argument, then the result
 of the previous step will be passed as the first argument of the function.
-Uf you did specify the `param_name` argument, then the result of the previous
+If you did specify the `param_name` argument, then the result of the previous
 step can be found in key word arguments with the param name you specified.
 
 You can add sequential steps with `.call_next` method of the pipeline.
 
+If you don't want to pass the result of the previous step to the next one,
+you can use `.call_after` method of the pipeline.
+
 ### Mapper step
 
 This step runs specified task for each item of the previous task's result spawning

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq-pipelines"
-version = "0.1.0"
+version = "0.1.1"
 description = "Taskiq pipelines for task chaining."
 authors = ["Pavel Kirilin <[email protected]>"]
 readme = "README.md"
@@ -27,7 +27,7 @@ keywords = ["taskiq", "pipelines", "tasks", "distributed", "async"]
 python = "^3.8.1"
 taskiq = ">=0.0.8, <1"
 typing-extensions = "^4.3.0"
-pydantic = "^1.6.2"
+pydantic = "^2"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7"

pyproject.toml

@@ -9,16 +9,16 @@
 class AbstractStep(ABC):
     """Abstract pipeline step."""
 
-    step_name: str
-    known_steps: "Dict[str, Type[AbstractStep]]" = {}
+    _step_name: str
+    _known_steps: "Dict[str, Type[AbstractStep]]" = {}
 
     def __init_subclass__(cls, step_name: str, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
         # Sets step name to the step.
-        cls.step_name = step_name
+        cls._step_name = step_name
         # Registers new subclass in the dict of
         # known steps.
-        cls.known_steps[step_name] = cls
+        cls._known_steps[step_name] = cls
 
     @abstractmethod
     def dumps(self) -> str:

taskiq_pipelines/abc.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/constants.py

@@ -42,7 +42,9 @@ async def post_save(  # noqa: C901, WPS212
             return
         pipeline_data = message.labels[PIPELINE_DATA]
         try:
-            steps_data = pydantic.parse_raw_as(List[DumpedStep], pipeline_data)
+            steps_data = pydantic.TypeAdapter(List[DumpedStep]).validate_json(
+                pipeline_data,
+            )
         except ValueError:
             return
         if current_step_num + 1 >= len(steps_data):
@@ -99,7 +101,7 @@ async def on_error(
             return
         pipe_data = message.labels[PIPELINE_DATA]
         try:
-            steps = pydantic.parse_raw_as(List[DumpedStep], pipe_data)
+            steps = pydantic.TypeAdapter(List[DumpedStep]).validate_json(pipe_data)
         except ValueError:
             return
         if current_step_num == len(steps) - 1:

taskiq_pipelines/middleware.py

@@ -1,13 +1,23 @@
 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
 from taskiq.decor import AsyncTaskiqDecoratedTask
 from taskiq.kicker import AsyncKicker
 from typing_extensions import ParamSpec
 
-from taskiq_pipelines.constants import CURRENT_STEP, PIPELINE_DATA
+from taskiq_pipelines.constants import CURRENT_STEP, EMPTY_PARAM_NAME, PIPELINE_DATA
 from taskiq_pipelines.steps import FilterStep, MapperStep, SequentialStep, parse_step
 
 _ReturnType = TypeVar("_ReturnType")
@@ -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,13 +104,14 @@ 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.
         """
         self.steps.append(
             DumpedStep(
-                step_type=SequentialStep.step_name,
+                step_type=SequentialStep._step_name,  # noqa: WPS437
                 step_data=SequentialStep.from_task(
                     task=task,
                     param_name=param_name,
@@ -111,6 +122,62 @@ def call_next(
         )
         return self
 
+    @overload
+    def call_after(
+        self: "Pipeline[_FuncParams, _ReturnType]",
+        task: Union[
+            AsyncKicker[Any, Coroutine[Any, Any, _T2]],
+            AsyncTaskiqDecoratedTask[Any, Coroutine[Any, Any, _T2]],
+        ],
+        **additional_kwargs: Any,
+    ) -> "Pipeline[_FuncParams, _T2]":
+        ...
+
+    @overload
+    def call_after(
+        self: "Pipeline[_FuncParams, _ReturnType]",
+        task: Union[
+            AsyncKicker[Any, _T2],
+            AsyncTaskiqDecoratedTask[Any, _T2],
+        ],
+        **additional_kwargs: Any,
+    ) -> "Pipeline[_FuncParams, _T2]":
+        ...
+
+    def call_after(
+        self,
+        task: Union[
+            AsyncKicker[Any, Any],
+            AsyncTaskiqDecoratedTask[Any, Any],
+        ],
+        **additional_kwargs: Any,
+    ) -> Any:
+        """
+        Adds sequential step.
+
+        This task will be executed right after
+        the previous and result of the previous task
+        is not passed to the next task.
+
+        This is equivalent to call_next(task, param_name=-1).
+
+        :param task: task to execute.
+        :param additional_kwargs: additional kwargs to task.
+        :return: updated pipeline.
+        """
+        self.steps.append(
+            DumpedStep(
+                step_type=SequentialStep._step_name,  # noqa: WPS437
+                step_data=SequentialStep.from_task(
+                    task=task,
+                    param_name=EMPTY_PARAM_NAME,
+                    **additional_kwargs,
+                ).dumps(),
+                task_id="",
+            ),
+        )
+        return self
+
     @overload
     def map(
         self: "Pipeline[_FuncParams, _ReturnType]",
@@ -169,7 +236,7 @@ def map(
         """
         self.steps.append(
             DumpedStep(
-                step_type=MapperStep.step_name,
+                step_type=MapperStep._step_name,  # noqa: WPS437
                 step_data=MapperStep.from_task(
                     task=task,
                     param_name=param_name,
@@ -241,7 +308,7 @@ def filter(
         """
         self.steps.append(
             DumpedStep(
-                step_type=FilterStep.step_name,
+                step_type=FilterStep._step_name,  # noqa: WPS437
                 step_data=FilterStep.from_task(
                     task=task,
                     param_name=param_name,
@@ -261,7 +328,7 @@ def dumps(self) -> str:
         :returns: serialized pipeline.
         """
         return json.dumps(
-            [step.dict() for step in self.steps],
+            [step.model_dump() for step in self.steps],
         )
 
     @classmethod
@@ -277,7 +344,7 @@ def loads(cls, broker: AsyncBroker, pipe_data: str) -> "Pipeline[Any, Any]":
         :return: new
         """
         pipe: "Pipeline[Any, Any]" = Pipeline(broker)
-        pipe.steps = pydantic.parse_raw_as(List[DumpedStep], pipe_data)
+        pipe.steps = pydantic.TypeAdapter(List[DumpedStep]).validate_json(pipe_data)
         return pipe
 
     async def kiq(