Repository: spetlr-org/spetlr (language: python)

Issue:
Add Privileged Role Administrator
the Privileged Role Administrator right is needed on the pipeline SPN to give Directory Reader rights to the sql server. This right is not created as part of the setup_deploying_SPN.ps1 scriprt. Add it.


Discussion / hints:

Fix patch:
diff --git a/docs/etl/README.md b/docs/etl/README.md
index d194be051..45f59ee0d 100644
--- a/docs/etl/README.md
+++ b/docs/etl/README.md
@@ -46,7 +46,7 @@ outputs. The special properties of each type are
  - **Transformer** handle a given number of inputs and adds a single result dataframe
  - **Loader** acts as a sink, while passing its input on to the next sink
 
-The **Transformer** can run in two modes, controlled by the flag consume_inputs that is True by default. When True the transformer conumes its inputs and adds the output to the result dataframe. In usecases when there is a need to keep previously extracted (or transformed) dataframes after a transformation step set consume_inputs to False. When working with non conuming transformers it is crucial to set dataset input keys and dataset output keys. This ensures that the transformer and/or loader has explicit information on which dataframe(s) to handle.
+The **Transformer** can run in two modes, controlled by the flag consume_inputs that is True by default. When True the transformer consumes its inputs and adds the output to the result dataframe. In usecases when there is a need to keep previously extracted (or transformed) dataframes after a transformation step, set consume_inputs to False. When working with non consuming transformers it is crucial to set dataset input keys and dataset output keys. This ensures that the transformer and/or loader has explicit information on which dataframe(s) to handle.
 
 The special case of the  **Orchestrator** is that it takes all its steps and executes them
 in sequence on its inputs. Running in the default `execute()` method, the inputs are empty,
@@ -55,6 +55,7 @@ but an orchestrator can also be added as part of another orchestrator with the `
 For the most general case of a many-to-many transformation, implement your step by inheriting
 from the `EtlBase` class.
 
+For simple transformers and loaders, where there is only one dataframe to process and load, it is possible to make the extractor or transformer stop the etl flow. If either the extractor or the transformer returns None as the result, the ETL process is stopped without triggering any errors. 
 
 ## Usage examples:
 
diff --git a/docs/spark/README.md b/docs/spark/README.md
index d3229ab60..2a8b2cfb8 100644
--- a/docs/spark/README.md
+++ b/docs/spark/README.md
@@ -1,7 +1,14 @@
 
 # Spark class
 
-The `Spark` class should be utilized to start a Spark session in Spetlr.
+The `Spark` class should be utilized to start a Spark session in Spetlr:
+
+```python 
+from spetlr.spark import Spark
+
+spark = Spark.get()
+```
+
 Sessions are created using the standard `pyspark.sql.SparkSession` class.
 However, it is also possible to switch to Databricks Connect for Python,
 and use the `databricks.connect.DatabricksSession` class instead. This can
@@ -28,7 +35,7 @@ Just set the correct host URL to your Databricks workspace, and change
 the profile name below to your liking. Databricks will let you choose
 a cluster and remember it. 
 
-``` 
+```sh
 $my_profile='profile1',
 $host_url='https://xxxxxxxxxx.azuredatabricks.net'
 
@@ -39,6 +46,6 @@ $host_url='https://xxxxxxxxxx.azuredatabricks.net'
 databricks auth login --configure-cluster --profile $my_profile --host $host_url
 
 databricks auth env --profile $my_profile
-``` 
+```
 
 [Additional info on Databricks Connect can be found here...](https://docs.databricks.com/en/dev-tools/databricks-connect/python/index.html)
diff --git a/src/spetlr/etl/loader.py b/src/spetlr/etl/loader.py
index f2b020903..291d59627 100644
--- a/src/spetlr/etl/loader.py
+++ b/src/spetlr/etl/loader.py
@@ -29,7 +29,9 @@ def __init__(
     def etl(self, inputs: dataset_group) -> dataset_group:
         if len(self.dataset_input_key_list) > 0:
             if len(self.dataset_input_key_list) == 1:
-                self.save(inputs[self.dataset_input_key_list[0]])
+                df = inputs[self.dataset_input_key_list[0]]
+                if df is not None:
+                    self.save(df)
             else:
                 datasetFilteret = {
                     datasetKey: df
@@ -38,7 +40,9 @@ def etl(self, inputs: dataset_group) -> dataset_group:
                 }
                 self.save_many(datasetFilteret)
         elif len(inputs) == 1:
-            self.save(next(iter(inputs.values())))
+            df = next(iter(inputs.values()))
+            if df is not None:
+                self.save(df)
         else:
             self.save_many(inputs)
 
diff --git a/src/spetlr/etl/transformer.py b/src/spetlr/etl/transformer.py
index 6bb9c27bf..00063e0c0 100644
--- a/src/spetlr/etl/transformer.py
+++ b/src/spetlr/etl/transformer.py
@@ -58,7 +58,9 @@ def etl(self, inputs: dataset_group) -> dataset_group:
 
         # Call process or process_many depending on the input_key count
         if len(dataset_input_keys) == 1:
-            df = self.process(inputs[dataset_input_keys[0]])
+            df = inputs[dataset_input_keys[0]]
+            if df is not None:
+                df = self.process(df)
         else:
             inputs_filtered = {
                 datasetKey: df
diff --git a/src/spetlr/spark.py b/src/spetlr/spark.py
index 182b48c80..45eda3960 100644
--- a/src/spetlr/spark.py
+++ b/src/spetlr/spark.py
@@ -88,7 +88,12 @@ def _get_db_connect(cls) -> Optional[SparkSession]:
                         spark.conf.set(key, value)
                 return spark
             except ImportError:
-                raise ValueError("databricks.connect not installed") from None
+                raise ValueError(
+                    "The 'databricks-connect' library is not installed.\n"
+                    "Install the library or set the 'SPETLR_DATABRICKS_CONNECT'\n"
+                    "environment variable to 'False'."
+                ) from None
+
         return None
 
     @classmethod


Test patch:
diff --git a/tests/local/eh/test_ehtodeltabronze_orchestrator.py b/tests/local/eh/test_ehtodeltabronze_orchestrator.py
index 76818ea00..31ebc1d0a 100644
--- a/tests/local/eh/test_ehtodeltabronze_orchestrator.py
+++ b/tests/local/eh/test_ehtodeltabronze_orchestrator.py
@@ -18,15 +18,16 @@
 
 class EhToDeltaBronzeOrchestratorTests(DataframeTestCase):
     def test_01_default(self):
+        empty_df = Spark.get().createDataFrame([], schema="Id INTEGER")
         eh_mock = Mock()
         dh_mock = Mock()
 
-        eh_mock.read = Mock(return_value=None)
+        eh_mock.read = Mock(return_value=empty_df)
 
-        with patch.object(EhJsonToDeltaExtractor, "read", return_value=None) as p1:
-            with patch.object(SimpleLoader, "save", return_value=None) as p2:
+        with patch.object(EhJsonToDeltaExtractor, "read", return_value=empty_df) as p1:
+            with patch.object(SimpleLoader, "save", return_value=empty_df) as p2:
                 with patch.object(
-                    EhToDeltaBronzeTransformer, "process", return_value=None
+                    EhToDeltaBronzeTransformer, "process", return_value=empty_df
                 ) as p3:
                     orchestrator = EhToDeltaBronzeOrchestrator(eh=eh_mock, dh=dh_mock)
                     orchestrator.execute()
@@ -55,12 +56,12 @@ def __init__(self):
                 )
                 self.filter_with(TestFilter())
 
-        eh_mock.read = Mock(return_value=None)
+        eh_mock.read = Mock(return_value=empty_df)
 
         with patch.object(EhJsonToDeltaExtractor, "read", return_value=empty_df) as p1:
-            with patch.object(SimpleLoader, "save", return_value=None) as p2:
+            with patch.object(SimpleLoader, "save", return_value=empty_df) as p2:
                 with patch.object(
-                    EhToDeltaBronzeTransformer, "process", return_value=None
+                    EhToDeltaBronzeTransformer, "process", return_value=empty_df
                 ) as p3:
                     with patch.object(
                         TestFilter, "process", return_value=empty_df
diff --git a/tests/local/etl/test_break_from_etl.py b/tests/local/etl/test_break_from_etl.py
new file mode 100644
index 000000000..36541a43a
--- /dev/null
+++ b/tests/local/etl/test_break_from_etl.py
@@ -0,0 +1,35 @@
+import unittest
+from typing import Optional
+
+from pyspark.sql import DataFrame
+
+from spetlr.etl import Extractor, Loader, Orchestrator, Transformer
+from spetlr.spark import Spark
+
+
+class BreakFromEtlTests(unittest.TestCase):
+    def test_without_break(self):
+        with self.assertRaises(NotImplementedError):
+            BreakTestOrchestrator(NoBreakExtractor()).execute()
+
+    def test_with_break(self):
+        BreakTestOrchestrator(BreakExtractor()).execute()
+        self.assertTrue(True)
+
+
+class NoBreakExtractor(Extractor):
+    def read(self) -> DataFrame:
+        return Spark.get().createDataFrame([], schema="Id INTEGER")
+
+
+class BreakExtractor(Extractor):
+    def read(self) -> Optional[DataFrame]:
+        return None
+
+
+class BreakTestOrchestrator(Orchestrator):
+    def __init__(self, extractor: Extractor):
+        super().__init__()
+        self.extract_from(extractor)
+        self.transform_with(Transformer())
+        self.load_into(Loader())