feat(workflow): add initial workflow and step classes

pylintrc

@@ -516,7 +516,7 @@ max-returns=6
 max-statements=50
 
 # Minimum number of public methods for a class (see R0903).
-min-public-methods=2
+min-public-methods=1
 
 
 [IMPORTS]

stepbystep/__init__.py

@@ -0,0 +1,150 @@
+"""
+Stepbystep
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any
+
+
+class StepModel:
+    """
+    A step model. It has a name that should be used by StepFactory to create the right step, and a specific
+    configuration.
+    """
+
+    def __init__(self, name: str, config: Dict[str, Any]):
+        self._name = name
+        self._config = config
+
+    @property
+    def name(self) -> str:
+        """
+        Name for this step model.
+        """
+        return self._name
+
+    @property
+    def config(self) -> Dict[str, Any]:
+        """
+        Configuration for this step model.
+        """
+        return self._config
+
+
+class WorkflowModel:
+    """
+    A model of workflow. It is composed of a list of StepModel instances.
+    """
+
+    def __init__(self, *steps: StepModel):
+        self._steps = list(steps)
+
+    @property
+    def steps(self) -> List[StepModel]:
+        """
+        List of steps that constitute this workflow model.
+        :return:
+        """
+        return self._steps
+
+
+class Step(ABC):
+    """
+    Base class to implement a step.
+    """
+
+    def __init__(self, model: StepModel, workflow: 'Workflow'):
+        self.model = model  # type: StepModel
+        self.workflow = workflow  # type: Workflow
+
+    def prepare(self):
+        """
+        Initial prepare task. Perform any automated operation before prompting the user.
+        """
+
+    def prompt(self):
+        """
+        Initial user prompt. Prompt the user for additional data that could be used by the run.
+        """
+
+    def prepare_before_run(self):
+        """
+        Prepare just before run. Perform any automated operation before prompting the user.
+        """
+
+    def prompt_before_run(self):
+        """
+        Prompt user just before run. Prompt the user for additional data that could be used by the run.
+        """
+
+    @abstractmethod
+    def run(self):
+        """
+        Main run. It should actually perform the task described by the step model.
+        """
+
+    def prompt_after_run(self):
+        """
+        Prompt user just after run.
+        """
+
+    def cleanup_after_run(self):
+        """
+        Perform cleanup operations just after run.
+        """
+
+    def cleanup(self):
+        """
+        Final cleanup.
+        """
+
+
+class StepFactory(ABC):
+    """
+    Step factory can create Step implementations for each supported StepModel.
+    """
+
+    @abstractmethod
+    def build_step(self, model: StepModel, workflow: 'Workflow') -> Step:
+        """
+        Build a step from it's model and the current workflow.
+        :param model: StepModel to create from
+        :param workflow: current workflow
+        :return: fresh Step instance, of None if this factory doesn't support provided StepModel
+        """
+
+
+class Workflow:
+    """
+    Workflow is able to run a defined WorkflowModel, using provided StepFactory.
+    """
+
+    def __init__(self, model: WorkflowModel, step_factory: StepFactory):
+        self.model = model
+        self.context = {}  # type: Dict[str, Any]
+        self.step_factory = step_factory
+
+    def run(self):
+        """
+        Run all steps declared in WorkflowModel.
+        """
+        steps = []  # type: List[Step]
+        for step_model in self.model.steps:
+            step = self.step_factory.build_step(step_model, self)
+            steps.append(step)
+
+        for step in steps:
+            step.prepare()
+
+        for step in steps:
+            step.prompt()
+
+        for step in steps:
+            step.prepare_before_run()
+            step.prompt_before_run()
+            step.run()
+            step.prompt_after_run()
+            step.cleanup_after_run()
+
+        for step in steps:
+            step.cleanup()

stepbystep_tests/test_workflow.py

@@ -0,0 +1,36 @@
+from stepbystep import Workflow, StepFactory, StepModel, Step, WorkflowModel
+
+
+def test_dummy_workflow():
+    dummy1_called_config = None
+    dummy2_called_config = None
+
+    dummy1_step_model = StepModel('dummy1', {'foo': 'bar1'})
+    dummy2_step_model = StepModel('dummy2', {'foo': 'bar2'})
+
+    class Dummy1Step(Step):
+        def run(self):
+            nonlocal dummy1_called_config
+            dummy1_called_config = self.model.config
+
+    class Dummy2Step(Step):
+        def run(self):
+            nonlocal dummy2_called_config
+            dummy2_called_config = self.model.config
+
+    class DummyStepFactory(StepFactory):
+        def build_step(self, model: StepModel, workflow: Workflow) -> Step:
+            if model.name == 'dummy1':
+                return Dummy1Step(model, workflow)
+            if model.name == 'dummy2':
+                return Dummy2Step(model, workflow)
+            return None
+
+    step_factory = DummyStepFactory()
+    workflow_model = WorkflowModel(dummy1_step_model, dummy2_step_model)
+
+    workflow = Workflow(workflow_model, step_factory)
+    workflow.run()
+
+    assert dummy1_called_config == {'foo': 'bar1'}
+    assert dummy2_called_config == {'foo': 'bar2'}