refactor: improve the logging around dve processing errors and align reporting to module name rather than legacy name

Author: georgeRobertson

src/dve/core_engine/exceptions.py

@@ -1,12 +1,8 @@
 """Exceptions emitted by the pipeline."""
 
-from collections.abc import Iterator
+import traceback
 from typing import Optional
 
-from dve.core_engine.backends.implementations.spark.types import SparkEntities
-from dve.core_engine.message import FeedbackMessage
-from dve.core_engine.type_hints import Messages
-
 
 class CriticalProcessingError(ValueError):
     """An exception emitted if critical errors are received."""
@@ -15,26 +11,18 @@ def __init__(
         self,
         error_message: str,
         *args: object,
-        messages: Optional[Messages],
-        entities: Optional[SparkEntities] = None
+        messages: Optional[list[str]] = None,
     ) -> None:
         super().__init__(error_message, *args)
         self.error_message = error_message
         """The error message explaining the critical processing error."""
         self.messages = messages
-        """The messages gathered at the time the error was emitted."""
-        self.entities = entities
-        """The entities as they exist at the time the error was emitted."""
-
-    @property
-    def critical_messages(self) -> Iterator[FeedbackMessage]:
-        """Critical messages which caused the processing error."""
-        yield from filter(lambda message: message.is_critical, self.messages)  # type: ignore
+        """The stacktrace for the messages."""
 
     @classmethod
     def from_exception(cls, exc: Exception):
         """Create from broader exception, for recording in processing errors"""
-        return cls(error_message=repr(exc), entities=None, messages=[])
+        return cls(error_message=repr(exc), messages=traceback.format_exception(exc))
 
 
 class EntityTypeMismatch(TypeError):

src/dve/reporting/utils.py

@@ -61,7 +61,7 @@ def dump_processing_errors(
     if not errors:
         raise AttributeError("errors list not passed")
 
-    error_file: URI = fh.joinuri(working_folder, "errors", "processing_errors.json")
+    error_file: URI = fh.joinuri(working_folder, "processing_errors", "processing_errors.json")
     processed = []
 
     for error in errors:
@@ -71,6 +71,7 @@ def dump_processing_errors(
                 "error_location": "processing",
                 "error_level": "integrity",
                 "error_message": error.error_message,
+                "error_traceback": error.messages,
             }
         )
 

tests/test_pipeline/test_foundry_ddb_pipeline.py

@@ -8,7 +8,7 @@
 import tempfile
 from uuid import uuid4
 
-import pytest
+import polars as pl
 
 from dve.core_engine.backends.implementations.duckdb.auditing import DDBAuditingManager
 from dve.core_engine.backends.implementations.duckdb.reference_data import DuckDBRefDataLoader
@@ -116,6 +116,42 @@ def test_foundry_runner_error(planet_test_files, temp_ddb_conn):
         output_loc, report_uri, audit_files = dve_pipeline.run_pipeline(sub_info)
         assert not fh.get_resource_exists(report_uri)
         assert not output_loc
+
+        perror_path = Path(
+            processing_folder,
+            sub_info.submission_id,
+            "processing_errors",
+            "processing_errors.json"
+        )
+        assert perror_path.exists()
+        perror_schema = {
+            "step_name": pl.Utf8(),
+            "error_location": pl.Utf8(),
+            "error_level": pl.Utf8(),
+            "error_message": pl.Utf8(),
+            "error_traceback": pl.List(pl.Utf8()),
+        }
+        expected_error_df = (
+            pl.DataFrame(
+                [
+                    {
+                        "step_name": "file_transformation",
+                        "error_location": "processing",
+                        "error_level": "integrity",
+                        "error_message": "ReaderLacksEntityTypeSupport()",
+                        "error_traceback": None,
+                    },
+                ],
+                perror_schema
+            )
+            .select(pl.col("step_name"), pl.col("error_location"), pl.col("error_message"))
+        )
+        actual_error_df = (
+            pl.read_json(perror_path, schema=perror_schema)
+            .select(pl.col("step_name"), pl.col("error_location"), pl.col("error_message"))
+        )
+        assert actual_error_df.equals(expected_error_df)
+
         assert len(list(fh.iter_prefix(audit_files))) == 2
 
 

tests/test_error_reporting/__init__.py tests/test_reporting/__init__.pytests/test_error_reporting/__init__.py renamed to tests/test_reporting/__init__.py

@@ -0,0 +1,51 @@
+"""test utility functions & objects in dve.reporting module"""
+
+import tempfile
+from pathlib import Path
+
+import polars as pl
+
+from dve.core_engine.exceptions import CriticalProcessingError
+from dve.reporting.utils import dump_processing_errors
+
+# pylint: disable=C0116
+
+
+def test_dump_processing_errors():
+    perror_schema = {
+        "step_name": pl.Utf8(),
+        "error_location": pl.Utf8(),
+        "error_level": pl.Utf8(),
+        "error_message": pl.Utf8(),
+        "error_stacktrace": pl.List(pl.Utf8()),
+    }
+    with tempfile.TemporaryDirectory() as temp_dir:
+        dump_processing_errors(
+            temp_dir,
+            "test_step",
+            [CriticalProcessingError("test error message")]
+        )
+
+        output_path = Path(temp_dir, "processing_errors")
+
+        assert output_path.exists()
+        assert len(list(output_path.iterdir())) == 1
+
+        expected_df = pl.DataFrame(
+            [
+                {
+                    "step_name": "test_step",
+                    "error_location": "processing",
+                    "error_level": "integrity",
+                    "error_message": "test error message",
+                    "error_stacktrace": None,
+                },
+            ],
+            perror_schema
+        )
+        error_df = pl.read_json(
+            Path(output_path, "processing_errors.json")
+        )
+        cols_to_check = ["step_name", "error_location", "error_level", "error_message"]
+
+        assert error_df.select(pl.col(k) for k in cols_to_check).equals(expected_df.select(pl.col(k) for k in cols_to_check))