fix: deal with pathing assumption that file had been moved to processed_file_path during file transformation

Author: georgeRobertson

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -1,6 +1,8 @@
 # pylint: disable=W0223
 """A duckdb pipeline for running on Foundry platform"""
 
+import shutil
+from pathlib import Path
 from typing import Optional
 
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
@@ -23,6 +25,15 @@
 class FoundryDDBPipeline(DDBDVEPipeline):
     """DuckDB pipeline for running on Foundry Platform"""
 
+    def _move_submission_to_processing_files_path(self, submission_info: SubmissionInfo):
+        """Move submitted file to 'processed_files_path'."""
+        _submitted_file_location = Path(
+            self._submitted_files_path, submission_info.file_name_with_ext  # type: ignore
+        )
+        _dest = Path(self.processed_files_path, submission_info.submission_id)
+        _dest.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(_submitted_file_location, _dest)
+
     def persist_audit_records(self, submission_info: SubmissionInfo) -> URI:
         """Write out key audit relations to parquet for persisting to datasets"""
         write_to = fh.joinuri(self.processed_files_path, submission_info.submission_id, "audit/")
@@ -113,8 +124,8 @@ def run_pipeline(
         try:
             sub_id: str = submission_info.submission_id
             report_uri = None
-            self._audit_tables.add_new_submissions(submissions=[submission_info])
-            self._audit_tables.mark_transform(submission_ids=[sub_id])
+            if self._submitted_files_path:
+                self._move_submission_to_processing_files_path(submission_info)
             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])

tests/test_pipeline/test_foundry_ddb_pipeline.py

@@ -5,6 +5,7 @@
 from datetime import datetime
 from pathlib import Path
 import shutil
+import tempfile
 from uuid import uuid4
 
 import pytest
@@ -116,3 +117,42 @@ def test_foundry_runner_error(planet_test_files, temp_ddb_conn):
         assert not fh.get_resource_exists(report_uri)
         assert not output_loc
         assert len(list(fh.iter_prefix(audit_files))) == 2
+
+
+def test_foundry_runner_with_submitted_files_path(movies_test_files, temp_ddb_conn):
+    db_file, conn = temp_ddb_conn
+    ref_db_file = Path(db_file.parent, "movies_refdata.duckdb").as_posix()
+    conn.sql(f"ATTACH '{ref_db_file}' AS movies_refdata")
+    conn.read_parquet(
+        get_test_file_path("movies/refdata/movies_sequels.parquet").as_posix()
+    ).to_table("movies_refdata.sequels")
+    processing_folder = Path(tempfile.mkdtemp()).as_posix()
+    submitted_files_path = Path(movies_test_files).as_posix()
+    sub_id = uuid4().hex
+    sub_info = SubmissionInfo(
+        submission_id=sub_id,
+        dataset_id="movies",
+        file_name="good_movies",
+        file_extension="json",
+        submitting_org="TEST",
+        datetime_received=datetime(2025,11,5)
+    )
+
+    DuckDBRefDataLoader.connection = conn
+    DuckDBRefDataLoader.dataset_config_uri = None
+
+    with DDBAuditingManager(db_file.as_uri(), None, conn) as audit_manager:
+        dve_pipeline = FoundryDDBPipeline(
+            processed_files_path=processing_folder,
+            audit_tables=audit_manager,
+            connection=conn,
+            rules_path=get_test_file_path("movies/movies_ddb.dischema.json").as_posix(),
+            submitted_files_path=submitted_files_path,
+            reference_data_loader=DuckDBRefDataLoader,
+        )
+        output_loc, report_uri, audit_files = dve_pipeline.run_pipeline(sub_info)
+
+        assert Path(processing_folder, sub_id, sub_info.file_name_with_ext).exists()
+        assert fh.get_resource_exists(report_uri)
+        assert len(list(fh.iter_prefix(output_loc))) == 2
+        assert len(list(fh.iter_prefix(audit_files))) == 3