Feat/live ingestion job summary

ingestify/__init__.py

@@ -8,7 +8,7 @@
     from .infra import retrieve_http
     from .source_base import Source, DatasetResource
     from .domain.models.fetch_policy import FetchPolicy
-    from .exceptions import StopProcessing
+    from .exceptions import StopProcessing, FatalError
     from .main import debug_source
 
 __version__ = "0.17.1"

ingestify/cmdline.py

@@ -7,7 +7,7 @@
 import click
 from dotenv import find_dotenv, load_dotenv
 
-from ingestify.exceptions import ConfigurationError, StopProcessing
+from ingestify.exceptions import ConfigurationError, StopProcessing, FatalError
 from ingestify.main import get_engine
 
 from ingestify import __version__
@@ -121,6 +121,9 @@ def run(
     except StopProcessing as e:
         logger.warning(f"Stopped early: {e}")
         sys.exit(e.exit_code)
+    except FatalError as e:
+        logger.error(f"Fatal error: {e}")
+        sys.exit(e.exit_code)
 
     logger.info("Done")
 

ingestify/domain/models/ingestion/ingestion_job.py

@@ -25,7 +25,7 @@
 )
 from ingestify.domain.models.dataset.dataset import DatasetLastModifiedAtMap
 from ingestify.domain.models.task.task_summary import TaskSummary, Operation
-from ingestify.exceptions import SaveError, IngestifyError, StopProcessing
+from ingestify.exceptions import SaveError, IngestifyError, StopProcessing, FatalError
 from ingestify.utils import TaskExecutor, chunker
 
 logger = logging.getLogger(__name__)
@@ -238,6 +238,11 @@ def __repr__(self):
 
 MAX_TASKS_PER_CHUNK = 10_000
 
+# How many additional tasks must accumulate before a live progress snapshot of
+# the IngestionJobSummary is persisted again. Keeps mid-run writes to roughly
+# one per PROGRESS_SAVE_INTERVAL tasks instead of one per (possibly tiny) batch.
+PROGRESS_SAVE_INTERVAL = 100
+
 
 class IngestionJob:
     def __init__(
@@ -249,6 +254,31 @@ def __init__(
         self.ingestion_job_id = ingestion_job_id
         self.ingestion_plan = ingestion_plan
         self.selector = selector
+        # task_count() at which the summary was last persisted mid-run.
+        self._last_progress_saved_at = 0
+
+    def _save_progress(
+        self,
+        store: DatasetStore,
+        ingestion_job_summary: IngestionJobSummary,
+        *,
+        force: bool = False,
+    ):
+        """Persist a live snapshot of the (still RUNNING) summary.
+
+        Only the parent row plus any FAILED task summaries are written (see
+        DatasetStore.save_ingestion_job_summary), so it is cheap enough to call
+        repeatedly. Throttled to roughly every PROGRESS_SAVE_INTERVAL tasks
+        unless ``force`` is set (used for the initial RUNNING row and when a new
+        chunk starts). Works for both the sync find_datasets flow and the async
+        submit/collect flow.
+        """
+        count = ingestion_job_summary.task_count()
+        if not force and count - self._last_progress_saved_at < PROGRESS_SAVE_INTERVAL:
+            return
+        self._last_progress_saved_at = count
+        ingestion_job_summary.recount()
+        store.save_ingestion_job_summary(ingestion_job_summary)
 
     def execute(
         self,
@@ -258,6 +288,11 @@ def execute(
     ) -> Iterator[IngestionJobSummary]:
         is_first_chunk = True
         ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self)
+        # Persist the RUNNING row up front so the job is observable from the
+        # moment it starts — and so a record survives even if the run never
+        # reaches its final yield (e.g. an async source that keeps polling).
+        self._last_progress_saved_at = 0
+        self._save_progress(store, ingestion_job_summary, force=True)
         # Process all items in batches. Yield a IngestionJobSummary per batch
 
         logger.info("Finding metadata")
@@ -302,6 +337,13 @@ def execute(
 
                 # We need to include the to_batches as that will start the generator
                 batches = to_batches(dataset_resources)
+        except FatalError as e:
+            # Persist the failure so it isn't lost, then abort — not swallowed as
+            # a skipped find_datasets. (StopProcessing keeps its own controlled
+            # stop and is not intercepted here.)
+            ingestion_job_summary.set_exception(e)
+            yield ingestion_job_summary
+            raise
         except ValidationError as e:
             # Make sure to pass this to the highest level as this means the Source is wrong
             if "Field required" in str(e):
@@ -343,6 +385,12 @@ def execute(
                         batch = next(batches)
                     except StopIteration:
                         break
+            except FatalError as e:
+                # Persist the failure, then abort, instead of swallowing it as a
+                # failed batch fetch.
+                ingestion_job_summary.set_exception(e)
+                yield ingestion_job_summary
+                raise
             except Exception as e:
                 logger.exception("Failed to fetch next batch")
 
@@ -453,6 +501,11 @@ def execute(
                         ingestion_job_summary.set_finished()
                         yield ingestion_job_summary
                         raise
+                    except FatalError as e:
+                        logger.error("Fatal error — saving summary and aborting")
+                        ingestion_job_summary.set_exception(e)
+                        yield ingestion_job_summary
+                        raise
 
                     ingestion_job_summary.add_task_summaries(results)
                 else:
@@ -462,13 +515,19 @@ def execute(
                     )
                 ingestion_job_summary.increase_skipped_tasks(skipped_tasks)
 
+            # Live snapshot after each batch (throttled) so the summary's
+            # counters/state stay current while the job runs.
+            self._save_progress(store, ingestion_job_summary)
+
             if ingestion_job_summary.task_count() >= MAX_TASKS_PER_CHUNK:
                 ingestion_job_summary.set_finished()
                 yield ingestion_job_summary
 
                 # Start a new one
                 is_first_chunk = False
                 ingestion_job_summary = IngestionJobSummary.new(ingestion_job=self)
+                self._last_progress_saved_at = 0
+                self._save_progress(store, ingestion_job_summary, force=True)
 
         if ingestion_job_summary.task_count() > 0 or is_first_chunk:
             # When there is interesting information to store, or there was no data at all, store it
@@ -496,6 +555,10 @@ def filtered_stream():
                             batch = next(batches)
                         except StopIteration:
                             return
+                except FatalError:
+                    # Propagate to the submit/collect handler (which persists the
+                    # summary); don't swallow it as a failed batch fetch.
+                    raise
                 except Exception as e:
                     logger.exception("Failed to fetch next batch")
                     ingestion_job_summary.set_exception(e)
@@ -556,12 +619,31 @@ def filtered_stream():
         resources = filtered_stream()
         done = False
 
-        while not done or source.has_pending():
-            done = source.submit(resources)
-
-            for dataset_resource in source.collect():
-                task_summary = self._store_async_result(dataset_resource, store)
-                ingestion_job_summary.add_task_summaries([task_summary])
+        # Unlike the sync path, submit/collect had no stop/fail handling: a
+        # StopProcessing (e.g. quota) or FatalError raised while collecting would
+        # propagate without ever persisting the summary, losing the record. Mirror
+        # the sync handler here so both are saved before aborting.
+        try:
+            while not done or source.has_pending():
+                done = source.submit(resources)
+
+                for dataset_resource in source.collect():
+                    task_summary = self._store_async_result(dataset_resource, store)
+                    ingestion_job_summary.add_task_summaries([task_summary])
+
+                    # Live snapshot (throttled) so long-running submit/collect jobs
+                    # are observable while polling, instead of only at the very end.
+                    self._save_progress(store, ingestion_job_summary)
+        except StopProcessing:
+            logger.info("StopProcessing raised — saving partial results and stopping")
+            ingestion_job_summary.set_finished()
+            yield ingestion_job_summary
+            raise
+        except FatalError as e:
+            logger.error("Fatal error — saving summary and aborting")
+            ingestion_job_summary.set_exception(e)
+            yield ingestion_job_summary
+            raise
 
         if ingestion_job_summary.task_count() > 0 or is_first_chunk:
             ingestion_job_summary.set_finished()

ingestify/domain/models/ingestion/ingestion_job_summary.py

@@ -70,7 +70,14 @@ def increase_skipped_tasks(self, skipped_tasks: int):
     def task_count(self):
         return len(self.task_summaries) + self.skipped_tasks
 
-    def _set_ended(self):
+    def recount(self):
+        """Recompute the task counters from the currently collected task
+        summaries, without ending the job or discarding any summaries.
+
+        Safe to call repeatedly while the job is still RUNNING so a live
+        snapshot (written mid-run) reflects up-to-date counts. ``_set_ended``
+        reuses this before it truncates ``task_summaries``.
+        """
         self.failed_tasks = len(
             [task for task in self.task_summaries if task.state == TaskState.FAILED]
         )
@@ -90,6 +97,9 @@ def _set_ended(self):
             + self.ignored_successful_tasks
             + self.skipped_tasks
         )
+
+    def _set_ended(self):
+        self.recount()
         self.ended_at = utcnow()
 
         # Only keep failed tasks. Rest isn't interesting
@@ -111,7 +121,9 @@ def set_skipped(self):
 
     @property
     def duration(self) -> timedelta:
-        return self.ended_at - self.started_at
+        # ended_at is None while the job is still RUNNING (live snapshots);
+        # measure against "now" so duration stays meaningful mid-run.
+        return (self.ended_at or utcnow()) - self.started_at
 
     def output_report(self):
         print(

ingestify/exceptions.py

@@ -26,3 +26,20 @@ class StopProcessing(IngestifyError):
     """
 
     exit_code = 2
+
+
+class FatalError(IngestifyError):
+    """Raised by a source or loader to signal that the run cannot continue and
+    must fail loudly. Unlike StopProcessing, this is NOT recoverable by
+    retrying later — e.g. a deactivated account or an otherwise misconfigured
+    source.
+
+    Already-processed datasets are preserved, but the run aborts with a
+    non-zero exit so schedulers surface it as a failure instead of a silent
+    success. It propagates out of find_datasets/collect rather than being
+    swallowed as a skipped find_datasets.
+
+    Exit code: 1.
+    """
+
+    exit_code = 1

ingestify/infra/store/dataset/sqlalchemy/repository.py

@@ -38,7 +38,7 @@
     DatasetCollectionMetadata,
 )
 from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobSummary
-from ingestify.domain.models.task.task_summary import TaskSummary
+from ingestify.domain.models.task.task_summary import TaskSummary, TaskState
 from ingestify.exceptions import IngestifyError
 from ingestify.utils import get_concurrency, key_from_dict
 
@@ -740,17 +740,26 @@ def next_identity(self):
 
     # TODO: consider moving the IngestionJobSummary methods to a different Repository
     def save_ingestion_job_summary(self, ingestion_job_summary: IngestionJobSummary):
+        """Upsert the summary row (keyed on ingestion_job_summary_id).
+
+        Only FAILED task summaries are persisted as child rows — successful,
+        ignored and skipped tasks are captured by the counters, not individual
+        rows. This keeps repeated live (mid-run) writes cheap (a successful run
+        writes no child rows at all) while still surfacing failures as soon as
+        they happen. The upsert is keyed on ingestion_job_summary_id, so live
+        snapshots update the same row in place.
+        """
         ingestion_job_summary_entities = [
             ingestion_job_summary.model_dump(exclude={"task_summaries"})
         ]
-        task_summary_entities = []
-        for task_summary in ingestion_job_summary.task_summaries:
-            task_summary_entities.append(
-                {
-                    **task_summary.model_dump(),
-                    "ingestion_job_summary_id": ingestion_job_summary.ingestion_job_summary_id,
-                }
-            )
+        task_summary_entities = [
+            {
+                **task_summary.model_dump(),
+                "ingestion_job_summary_id": ingestion_job_summary.ingestion_job_summary_id,
+            }
+            for task_summary in ingestion_job_summary.task_summaries
+            if task_summary.state == TaskState.FAILED
+        ]
 
         with self.session_provider.engine.connect() as connection:
             try: