py: add Pipeline.errors: return list of all errors

abhizer · abhizer · commit 69f28c42e8b7 · 2025-05-03T12:16:35.000Z
Signed-off-by: Abhinav Gyawali &lt;22275402+abhizer@users.noreply.github.com&gt;
diff --git a/python/docs/examples.rst b/python/docs/examples.rst
@@ -83,6 +83,36 @@ Starting a Pipeline
 
     pipeline.start()
 
+Analyzing Existing Feldera Pipeline for Errors
+==============================================
+
+First let's create a Feldera pipeline that errors from the web console, with the
+name ``check_error`` and invalid SQL as follows:
+
+.. code-block:: sql
+
+   SELECT invalid
+
+This will fail to compile.
+
+We can use this Python SDK to connect to this Feldera pipeline to check if it has
+any errors as follows:
+
+.. code-block:: python
+
+    pipeline = Pipeline.get("check_error", client)
+    err = pipeline.errors()
+
+    if len(err) != 0:
+        print("got err: ", err)
+
+Here, ``err`` is a list of all errors in this pipeline. The above code will emit
+the following output:
+
+.. code-block:: text
+
+   got err:  [{'sql_compilation': {'exit_code': 1, 'messages': [{'start_line_number': 1, 'start_column': 1, 'end_line_number': 1, 'end_column': 14, 'warning': False, 'error_type': 'Not supported', 'message': 'Raw \'SELECT\' statements are not supported; did you forget to CREATE VIEW?: SELECT "invalid"', 'snippet': '    1|SELECT invalid\n      ^^^^^^^^^^^^^^\n'}]}}]
+
 
 Using Pandas DataFrames
 =======================
diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
@@ -816,3 +816,18 @@ def program_error(self) -> Mapping[str, Any]:
 
         self.refresh()
         return self._inner.program_error
+
+    def errors(self) -> List[Mapping[str, Any]]:
+        """
+        Returns a list of all errors in this pipeline.
+        """
+        errors = []
+        perr = self.program_error()
+        for e in perr.keys():
+            err = perr.get(e)
+            if err and err.get("exit_code", 0) != 0:
+                errors.append({e: err})
+        derr = self.deployment_error()
+        if derr:
+            errors.append(derr)
+        return errors
diff --git a/python/tests/test_pipeline_builder.py b/python/tests/test_pipeline_builder.py
@@ -1173,6 +1173,19 @@ def test_program_error1(self):
         pipeline.shutdown()
         pipeline.delete()
 
+    def test_errors0(self):
+        sql = "SELECT invalid"
+        name = "test_errors0"
+
+        try:
+            _ = PipelineBuilder(TEST_CLIENT, name, sql).create_or_replace()
+        except Exception:
+            pass
+
+        pipeline = Pipeline.get(name, TEST_CLIENT)
+
+        assert pipeline.errors()[0]["sql_compilation"]["exit_code"] != 0
+
 
 if __name__ == "__main__":
     unittest.main()
