Reset _has_been_called to False ahead of pipeline.connect

src/zenml/pipelines/base_pipeline.py

@@ -235,8 +235,14 @@ def steps(self, steps: Dict[str, BaseStep]) -> NoReturn:
         """
         raise PipelineInterfaceError("Cannot set steps manually!")
 
-    # TODO [ENG-376]: Enable specifying runtime configuration options either using
-    #  **kwargs here or by passing a `RuntimeConfiguration` object or a
+    def _reset_step_flags(self) -> None:
+        """Reset the _has_been_called flag at the beginning of a pipeline run,
+        to make sure a pipeline instance can be called more than once."""
+        for step in self.steps.values():
+            step._has_been_called = False
+
+    # TODO [ENG-376]: Enable specifying runtime configuration options either
+    #  using **kwargs here or by passing a `RuntimeConfiguration` object or a
     #  path to a config file.
     def run(
         self,
@@ -287,6 +293,8 @@ def run(
             },
         )
 
+        self._reset_step_flags()
+
         return stack.deploy_pipeline(
             self, runtime_configuration=runtime_configuration
         )

tests/unit/pipelines/test_base_pipeline.py

@@ -240,3 +240,16 @@ def test_setting_pipeline_parameter_name_when_initializing_pipeline(
     assert step_instance.pipeline_parameter_name is None
     one_step_pipeline(step_instance)
     assert step_instance.pipeline_parameter_name == "step_"
+
+
+def test_calling_a_pipeline_twice_raises_no_exception(
+    one_step_pipeline, empty_step
+):
+    """Tests that calling one pipeline instance twice does not raise
+    any exception."""
+
+    pipeline_instance = one_step_pipeline(empty_step())
+
+    with does_not_raise():
+        pipeline_instance.run()
+        pipeline_instance.run()