Sequence groupings

CHANGELOG.md

@@ -3,28 +3,35 @@
 ## [Unreleased]
 
 ### Added
+
 - `ConsoleMetricsExporter` for locally debugging pipelines without an APM service. (#46)
 - Built-in Elastic APM metrics exporter. (#47)
+- `Sequence` class for logical groupings of steps. (#53)
 
 ### Changed
+
 - Bumped `redis` dependency to 4.2.2. (#50)
-- Dependencies for `MetricsExporter` and `PipelineLock` modules must now be installed separately through `watergrid[...]` metapackages. (#54)
 - Bumped `elastic-apm` to 6.9.1. (#56)
 
 ### Deprecated
 
 ### Removed
 
+- Dependencies for `MetricsExporter` and `PipelineLock` modules are no longer included in the base package and must
+now be installed separately through `watergrid[...]` metapackages. (#54)
+
 ### Fixed
 
 ### Security
 
 ## [1.0.1] - 2022-04-01
 ### Fixed
+
 - Broken release pipeline to PyPI.
 
 ## [1.0.0] - 2022-04-01
 ### Added
+
 - Core library functionality.
 - High availability pipeline using the `HAPipeline` class.
 - `RedisLock` for high availability mode using Redis.
@@ -34,6 +41,7 @@
 - Base class for building a metric exporter to an external APM solution.
 
 ### Fixed
+
 - Staggered node startup no longer causes mid-interval pipeline runs on other nodes in HA mode.
 
 

test/sequence_tests.py

@@ -0,0 +1,38 @@
+import unittest
+
+from watergrid.context import DataContext
+from watergrid.pipelines.pipeline import Pipeline
+from watergrid.steps import Sequence
+from watergrid.steps import Step
+
+
+class MockStep(Step):
+    def __init__(self):
+        super().__init__("mock_step")
+        self.flag = False
+
+    def run(self, context: DataContext):
+        self.flag = True
+
+    def get_flag(self) -> bool:
+        return self.flag
+
+
+class TestSequence(Sequence):
+    def __init__(self):
+        super().__init__("test_sequence")
+
+
+class SequenceTestCase(unittest.TestCase):
+    def test_sequence_adds_steps(self):
+        pipeline = Pipeline("test_pipeline")
+        mock_step = MockStep()
+        sequence = TestSequence()
+        sequence.add_step(mock_step)
+        pipeline.add_steps(sequence)
+        pipeline.run()
+        self.assertTrue(mock_step.get_flag())
+
+
+if __name__ == "__main__":
+    unittest.main()

watergrid/pipelines/pipeline.py

@@ -6,6 +6,7 @@
 from watergrid.context import DataContext, OutputMode, ContextMetadata
 from watergrid.metrics.MetricsStore import MetricsStore
 from watergrid.pipelines.pipeline_verifier import PipelineVerifier
+from watergrid.steps import Sequence
 from watergrid.steps import Step
 
 
@@ -32,6 +33,15 @@ def add_step(self, step: Step):
         """
         self._steps.append(step)
 
+    def add_steps(self, steps: Sequence):
+        """
+        Adds a sequence of steps to the pipeline.
+        :param steps: Sequence implementation that contains the steps to add.
+        :return: None
+        """
+        for step in steps.export_steps():
+            self.add_step(step)
+
     def run(self):
         """
         Blocking operation that runs all steps in the pipeline once.

watergrid/steps/__init__.py

@@ -1 +1,2 @@
 from watergrid.steps.step import Step
+from watergrid.steps.sequence import Sequence

watergrid/steps/sequence.py

@@ -0,0 +1,31 @@
+from abc import ABC
+
+from watergrid.steps import Step
+
+
+class Sequence(ABC):
+    """
+    A sequence object is used to form a logical grouping of steps. It can be
+    used to simplify your pipeline initialization code, or it can be used to
+    rapidly re-use multiple steps at once.
+    """
+
+    def __init__(self, name: str):
+        self.name = name
+        self.steps = []
+
+    def add_step(self, step: Step) -> None:
+        """
+        Adds a step to the sequence.
+        :param step: Step to add.
+        :return: None
+        """
+        self.steps.append(step)
+
+    def export_steps(self) -> list:
+        """
+        Returns a list of all steps in the sequence. Used internally by the
+        pipeline class.
+        :return: List of all steps.
+        """
+        return self.steps