style: Ran black, isort, pylint and mypy. Fixes made to resolve linting and type checking issues. Added test for new get_submission_status pipeline method

Author: stevenhsd

docs/README.md

@@ -234,10 +234,10 @@ audit_manager = SparkAuditingManager(
 
 # Setting up the Pipeline (in this case the Spark implemented one)
 pipeline = SparkDVEPipeline(
+    processed_files_path="path/where/my/processed_files/should_go/",
     audit_tables=audit_manager,
     job_run_id=1,
     rules_path="path/to/my_dischema",
-    processed_files_path="path/where/my/processed_files/should_go/",
     submitted_files_path="path/to/my/cwt_files/",
     reference_data_loader=SparkParquetRefDataLoader,
     spark=spark

src/dve/core_engine/backends/base/auditing.py

@@ -383,7 +383,7 @@ def mark_failed(self, submissions: list[str], **kwargs):
                 submission_id=submission_id,
                 processing_status="failed",
                 submission_result="processing_failed",
-                **kwargs
+                **kwargs,
             )
             for submission_id in submissions
         ]
@@ -497,26 +497,32 @@ def get_submission_statistics(self, submission_id: str) -> Optional[SubmissionSt
             )
         except StopIteration:
             return None
+
     def get_submission_status(self, submission_id: str) -> Optional[SubmissionStatus]:
         """Get the latest submission status for a submission"""
-        
+
         try:
-            processing_rec: ProcessingStatusRecord = next(self._processing_status.conv_to_records(
-                self._processing_status.get_most_recent_records(
-                    order_criteria=[OrderCriteria("time_updated", True)],
-                    pre_filter_criteria=[FilterCriteria("submission_id", submission_id)]
-                    )))
+            processing_rec: ProcessingStatusRecord = next(  # type: ignore
+                self._processing_status.conv_to_records(
+                    self._processing_status.get_most_recent_records(
+                        order_criteria=[OrderCriteria("time_updated", True)],
+                        pre_filter_criteria=[FilterCriteria("submission_id", submission_id)],
+                    )
+                )
+            )
         except StopIteration:
             return None
         sub_status = SubmissionStatus()
-        sub_stats_rec: Optional[SubmissionStatisticsRecord] = self.get_submission_statistics(submission_id)
+        sub_stats_rec: Optional[SubmissionStatisticsRecord] = self.get_submission_statistics(
+            submission_id
+        )
         if processing_rec.submission_result == "processing_failed":
             sub_status.processing_failed = True
         if processing_rec.submission_result == "validation_failed":
             sub_status.validation_failed = True
         if sub_stats_rec:
             sub_status.number_of_records = sub_stats_rec.record_count
-        
+
         return sub_status
 
     def __enter__(self):

src/dve/core_engine/backends/utilities.py

@@ -1,20 +1,18 @@
 """Necessary, otherwise uncategorised backend functionality."""
 
-import json
 import sys
 from dataclasses import is_dataclass
 from datetime import date, datetime, time
 from decimal import Decimal
 from typing import GenericAlias  # type: ignore
-from typing import Any, ClassVar, Optional, Union
+from typing import Any, ClassVar, Union
 
 import polars as pl  # type: ignore
 from polars.datatypes.classes import DataTypeClass as PolarsType
 from pydantic import BaseModel, create_model
 
-import dve.parser.file_handling as fh
 from dve.core_engine.backends.base.utilities import _get_non_heterogenous_type
-from dve.core_engine.type_hints import URI, Messages
+from dve.core_engine.type_hints import Messages
 
 # We need to rely on a Python typing implementation detail in Python <= 3.7.
 if sys.version_info[:2] <= (3, 7):

src/dve/core_engine/exceptions.py

@@ -1,6 +1,7 @@
 """Exceptions emitted by the pipeline."""
 
 from collections.abc import Iterator
+from typing import Optional
 
 from dve.core_engine.backends.implementations.spark.types import SparkEntities
 from dve.core_engine.message import FeedbackMessage
@@ -11,7 +12,11 @@ class CriticalProcessingError(ValueError):
     """An exception emitted if critical errors are received."""
 
     def __init__(
-        self, error_message: str, *args: object, messages: Messages, entities: SparkEntities
+        self,
+        error_message: str,
+        *args: object,
+        messages: Optional[Messages],
+        entities: Optional[SparkEntities] = None
     ) -> None:
         super().__init__(error_message, *args)
         self.error_message = error_message
@@ -24,13 +29,13 @@ def __init__(
     @property
     def critical_messages(self) -> Iterator[FeedbackMessage]:
         """Critical messages which caused the processing error."""
-        yield from filter(lambda message: message.is_critical, self.messages)
-    
+        yield from filter(lambda message: message.is_critical, self.messages)  # type: ignore
+
     @classmethod
-    def from_exception(cls, exc:Exception):
-        return cls(error_message = repr(exc),
-                   entities=None,
-                   messages=[])
+    def from_exception(cls, exc: Exception):
+        """Create from broader exception, for recording in processing errors"""
+        return cls(error_message=repr(exc), entities=None, messages=[])
+
 
 class EntityTypeMismatch(TypeError):
     """An exception emitted if entity type outputs from two collaborative objects are different."""

src/dve/core_engine/templating.py

@@ -11,12 +11,13 @@
 from dve.core_engine.type_hints import JSONable, TemplateVariables
 
 
-class PreserveTemplateUndefined(jinja2.Undefined):
+class PreserveTemplateUndefined(jinja2.Undefined):  # pylint: disable=too-few-public-methods
     """
-    Preserve the original template in instances where the value cannot be populated. Whilst this 
+    Preserve the original template in instances where the value cannot be populated. Whilst this
     may result in templates coming back in the FeedbackMessage object, it's more useful to know
     exactly what should have been populated rather than just returning blank values.
     """
+
     def __str__(self):
         return "{{" + self._undefined_name + "}}"
 

src/dve/pipeline/duckdb_pipeline.py

@@ -23,21 +23,21 @@ class DDBDVEPipeline(BaseDVEPipeline):
 
     def __init__(
         self,
+        processed_files_path: URI,
         audit_tables: DDBAuditingManager,
         connection: DuckDBPyConnection,
         rules_path: Optional[URI],
-        processed_files_path: Optional[URI],
         submitted_files_path: Optional[URI],
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
         job_run_id: Optional[int] = None,
     ):
         self._connection = connection
         super().__init__(
+            processed_files_path,
             audit_tables,
             DuckDBDataContract(connection=self._connection),
             DuckDBStepImplementations.register_udfs(connection=self._connection),
             rules_path,
-            processed_files_path,
             submitted_files_path,
             reference_data_loader,
             job_run_id,

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -1,17 +1,23 @@
+# pylint: disable=W0223
 """A duckdb pipeline for running on Foundry platform"""
 
 from typing import Optional
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_get_entity_count, duckdb_write_parquet
+
+from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_get_entity_count,
+    duckdb_write_parquet,
+)
 from dve.core_engine.exceptions import CriticalProcessingError
 from dve.core_engine.models import SubmissionInfo
-from dve.core_engine.type_hints import URI, Failed
+from dve.core_engine.type_hints import URI
+from dve.parser import file_handling as fh
 from dve.parser.file_handling.implementations.file import LocalFilesystemImplementation
 from dve.parser.file_handling.service import _get_implementation
 from dve.pipeline.duckdb_pipeline import DDBDVEPipeline
 from dve.pipeline.utils import SubmissionStatus
-from dve.parser import file_handling as fh
 from dve.reporting.utils import dump_processing_errors
 
+
 @duckdb_get_entity_count
 @duckdb_write_parquet
 class FoundryDDBPipeline(DDBDVEPipeline):
@@ -24,12 +30,12 @@ def persist_audit_records(self, submission_info: SubmissionInfo) -> URI:
             write_to = fh.file_uri_to_local_path(write_to)
             write_to.parent.mkdir(parents=True, exist_ok=True)
             write_to = write_to.as_posix()
-        self.write_parquet(
-            self._audit_tables._processing_status.get_relation(),
+        self.write_parquet(  # type: ignore # pylint: disable=E1101
+            self._audit_tables._processing_status.get_relation(),  # pylint: disable=W0212
             fh.joinuri(write_to, "processing_status.parquet"),
         )
-        self.write_parquet(
-            self._audit_tables._submission_statistics.get_relation(),
+        self.write_parquet(  # type: ignore # pylint: disable=E1101
+            self._audit_tables._submission_statistics.get_relation(),  # pylint: disable=W0212
             fh.joinuri(write_to, "submission_statistics.parquet"),
         )
         return write_to
@@ -39,62 +45,70 @@ def file_transformation(
     ) -> tuple[SubmissionInfo, SubmissionStatus]:
         try:
             return super().file_transformation(submission_info)
-        except Exception as exc: # pylint: disable=W0718
+        except Exception as exc:  # pylint: disable=W0718
             self._logger.error(f"File transformation raised exception: {exc}")
             self._logger.exception(exc)
             dump_processing_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "file_transformation",
-                    [CriticalProcessingError.from_exception(exc)]
-                )
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "file_transformation",
+                [CriticalProcessingError.from_exception(exc)],
+            )
             self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
 
-    def apply_data_contract(self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None) -> tuple[SubmissionInfo | SubmissionStatus]:
+    def apply_data_contract(
+        self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
+    ) -> tuple[SubmissionInfo, SubmissionStatus]:
         try:
             return super().apply_data_contract(submission_info, submission_status)
-        except Exception as exc: # pylint: disable=W0718
+        except Exception as exc:  # pylint: disable=W0718
             self._logger.error(f"Apply data contract raised exception: {exc}")
             self._logger.exception(exc)
             dump_processing_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "contract",
-                    [CriticalProcessingError.from_exception(exc)]
-                )
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "contract",
+                [CriticalProcessingError.from_exception(exc)],
+            )
             self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
 
-    def apply_business_rules(self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None):
+    def apply_business_rules(
+        self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
+    ):
         try:
             return super().apply_business_rules(submission_info, submission_status)
-        except Exception as exc: # pylint: disable=W0718
+        except Exception as exc:  # pylint: disable=W0718
             self._logger.error(f"Apply business rules raised exception: {exc}")
             self._logger.exception(exc)
             dump_processing_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "business_rules",
-                    [CriticalProcessingError.from_exception(exc)]
-                )
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "business_rules",
+                [CriticalProcessingError.from_exception(exc)],
+            )
             self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
-    
-    def error_report(self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None):
+
+    def error_report(
+        self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
+    ):
         try:
             return super().error_report(submission_info, submission_status)
-        except Exception as exc: # pylint: disable=W0718
+        except Exception as exc:  # pylint: disable=W0718
             self._logger.error(f"Error reports raised exception: {exc}")
             self._logger.exception(exc)
             sub_stats = None
             report_uri = None
             dump_processing_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "error_report",
-                    [CriticalProcessingError.from_exception(exc)]
-                )
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "error_report",
+                [CriticalProcessingError.from_exception(exc)],
+            )
             self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, submission_status, sub_stats, report_uri
 
-    def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI], Optional[URI], URI]:
+    def run_pipeline(
+        self, submission_info: SubmissionInfo
+    ) -> tuple[Optional[URI], Optional[URI], URI]:
         """Sequential single submission pipeline runner"""
         try:
             sub_id: str = submission_info.submission_id
@@ -104,8 +118,12 @@ def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI],
             sub_info, sub_status = self.file_transformation(submission_info=submission_info)
             if not (sub_status.validation_failed or sub_status.processing_failed):
                 self._audit_tables.mark_data_contract(submission_ids=[sub_id])
-                sub_info, sub_status = self.apply_data_contract(submission_info=sub_info, submission_status=sub_status)
-                self._audit_tables.mark_business_rules(submissions=[(sub_id, sub_status.validation_failed)])
+                sub_info, sub_status = self.apply_data_contract(
+                    submission_info=sub_info, submission_status=sub_status
+                )
+                self._audit_tables.mark_business_rules(
+                    submissions=[(sub_id, sub_status.validation_failed)]
+                )
                 sub_info, sub_status = self.apply_business_rules(
                     submission_info=submission_info, submission_status=sub_status
                 )
@@ -118,15 +136,15 @@ def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI],
                     submission_info=submission_info, submission_status=sub_status
                 )
                 self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
-        except Exception as err: # pylint: disable=W0718
+        except Exception as err:  # pylint: disable=W0718
             self._logger.error(
-                f"During processing of submission_id: {sub_id}, the following exception was raised: {err}"
+                f"During processing of submission_id: {sub_id}, this exception was raised: {err}"
             )
             dump_processing_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "run_pipeline",
-                    [CriticalProcessingError.from_exception(err)]
-                )
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "run_pipeline",
+                [CriticalProcessingError.from_exception(err)],
+            )
             self._audit_tables.mark_failed(submissions=[sub_id])
         finally:
             audit_files_uri = self.persist_audit_records(submission_info=submission_info)