feat: initial work to add Foundry pipeline for running DVE

Author: stevenhsd

src/dve/pipeline/duckdb_pipeline.py

@@ -24,23 +24,23 @@ class DDBDVEPipeline(BaseDVEPipeline):
     def __init__(
         self,
         audit_tables: DDBAuditingManager,
-        job_run_id: int,
         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__(
             audit_tables,
-            job_run_id,
             DuckDBDataContract(connection=self._connection),
             DuckDBStepImplementations.register_udfs(connection=self._connection),
             rules_path,
             processed_files_path,
             submitted_files_path,
             reference_data_loader,
+            job_run_id
         )
 
     # pylint: disable=arguments-differ
@@ -50,3 +50,4 @@ def write_file_to_parquet(  # type: ignore
         return super().write_file_to_parquet(
             submission_file_uri, submission_info, output, DuckDBPyRelation
         )
+    

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -0,0 +1,43 @@
+"""A duckdb pipeline for running on Foundry platform"""
+from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_write_parquet
+from dve.core_engine.models import SubmissionInfo
+from dve.pipeline.duckdb_pipeline import DDBDVEPipeline
+from dve.pipeline.utils import SubmissionStatus
+from dve.parser import file_handling as fh
+
+@duckdb_write_parquet
+class FoundryDDBPipeline(DDBDVEPipeline):
+    """DuckDB pipeline for running on Foundry Platform"""
+    def persist_audit_records(self, submission_info: SubmissionInfo):
+        """Write out key audit relations to parquet for persisting to datasets"""
+        write_to = fh.joinuri(self.processed_files_path, submission_info.submission_id, "audit/")
+        self.write_parquet(
+            self._audit_tables._processing_status.get_relation(),
+            write_to + "processing_status.parquet")
+        self.write_parquet(
+            self._audit_tables._submission_statistics.get_relation(),
+            write_to + "submission_statistics.parquet")
+    
+    def run_pipeline(self, submission_info: SubmissionInfo):
+        """Sequential single submission pipeline runner"""
+        try:
+            sub_id: str = submission_info.submission_id
+            self._audit_tables.add_new_submissions(submissions=[submission_info])
+            self._audit_tables.mark_transform(submission_ids=[sub_id])
+            sub_info = self.file_transformation(submission_info=submission_info)
+            if isinstance(sub_info, SubmissionInfo):
+                self._audit_tables.mark_data_contract(submission_ids=[sub_id])
+                sub_info, failed = self.apply_data_contract(submission_info=submission_info)
+                self._audit_tables.mark_business_rules(submissions=[(sub_info, failed)])
+                sub_info, sub_status = self.apply_business_rules(submission_info=submission_info, failed= failed)
+            else:
+                sub_status = SubmissionStatus(failed=True)    
+            self._audit_tables.mark_error_report(submissions=[(sub_id, sub_status.submission_result)])
+            sub_info, sub_status, sub_stats = self.error_report(submission_info=submission_info)
+            self._audit_tables.add_submission_statistics_records(subs_stats=[sub_stats])
+        except Exception as err:
+            self._logger.error(f"During processing of submission_id: {sub_id}, the following exception was raised: {err}")
+            self._audit_tables.mark_failed(submissions=[sub_id])
+        finally:
+            self.persist_audit_records(submission_info=submission_info)
+            

src/dve/pipeline/pipeline.py

@@ -44,13 +44,13 @@ class BaseDVEPipeline:
     def __init__(
         self,
         audit_tables: BaseAuditingManager,
-        job_run_id: int,
         data_contract: BaseDataContract,
         step_implementations: Optional[BaseStepImplementations[EntityType]],
         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._submitted_files_path = submitted_files_path
         self._processed_files_path = processed_files_path

src/dve/pipeline/spark_pipeline.py

@@ -26,23 +26,23 @@ class SparkDVEPipeline(BaseDVEPipeline):
     def __init__(
         self,
         audit_tables: SparkAuditingManager,
-        job_run_id: int,
         rules_path: Optional[URI],
         processed_files_path: Optional[URI],
         submitted_files_path: Optional[URI],
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
         spark: Optional[SparkSession] = None,
+        job_run_id: Optional[int] = None,
     ):
         self._spark = spark if spark else SparkSession.builder.getOrCreate()
         super().__init__(
             audit_tables,
-            job_run_id,
             SparkDataContract(spark_session=self._spark),
             SparkStepImplementations.register_udfs(self._spark),
             rules_path,
             processed_files_path,
             submitted_files_path,
             reference_data_loader,
+            job_run_id
         )
 
     # pylint: disable=arguments-differ

tests/test_pipeline/test_duckdb_pipeline.py

@@ -4,6 +4,7 @@
 
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+import shutil
 from typing import Dict, Tuple
 from uuid import uuid4
 
@@ -16,6 +17,7 @@
 from dve.core_engine.models import SubmissionInfo
 import dve.parser.file_handling as fh
 from dve.pipeline.duckdb_pipeline import DDBDVEPipeline
+from dve.pipeline.foundry_ddb_pipeline import FoundryDDBPipeline
 
 from ..conftest import get_test_file_path
 from ..fixtures import temp_ddb_conn  # pylint: disable=unused-import
@@ -204,3 +206,32 @@ def test_error_report_step(
 
     audit_result = audit_manager.get_current_processing_info(submitted_file_info.submission_id)
     assert audit_result.processing_status == "success"
+
+def test_foundry_runner_success(planet_test_files, temp_ddb_conn):
+    db_file, conn = temp_ddb_conn
+    processing_folder = planet_test_files
+
+    DuckDBRefDataLoader.connection = conn
+    DuckDBRefDataLoader.dataset_config_uri = fh.get_parent(PLANETS_RULES_PATH)
+    sub_id = uuid4().hex
+    sub_info = SubmissionInfo.from_metadata_file(submission_id=sub_id,
+                                                 metadata_uri=PLANETS_RULES_PATH)
+    
+    shutil.copytree()
+
+    with DDBAuditingManager(db_file.as_uri(), None, conn) as audit_manager:
+        dve_pipeline = FoundryDDBPipeline(
+            audit_tables=audit_manager,
+            connection=conn,
+            rules_path=PLANETS_RULES_PATH,
+            processed_files_path=processing_folder,
+            submitted_files_path=None,
+            reference_data_loader=DuckDBRefDataLoader,
+        )
+        
+
+def test_foundry_runner_fail():
+    pass
+
+def test_foundry_runner_error():
+    pass