diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 8798985..94e59ab 100644 --- a/ingestify/__init__.py +++ b/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" diff --git a/ingestify/cmdline.py b/ingestify/cmdline.py index b432eda..5b4dffb 100644 --- a/ingestify/cmdline.py +++ b/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") diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index c76ee7c..b51dae3 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/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__) @@ -302,6 +302,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 +350,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 +466,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: @@ -496,6 +514,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 +578,27 @@ 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]) + 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() diff --git a/ingestify/exceptions.py b/ingestify/exceptions.py index cd311a9..011e989 100644 --- a/ingestify/exceptions.py +++ b/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 diff --git a/ingestify/tests/test_fatal_error.py b/ingestify/tests/test_fatal_error.py new file mode 100644 index 0000000..b95e615 --- /dev/null +++ b/ingestify/tests/test_fatal_error.py @@ -0,0 +1,181 @@ +"""Tests for FatalError exception.""" +from typing import Iterator +from unittest.mock import patch + +import pytest + +from ingestify import Source, DatasetResource +from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector +from ingestify.domain.models.fetch_policy import FetchPolicy +from ingestify.domain.models.ingestion.ingestion_job_summary import IngestionJobState +from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan +from ingestify.exceptions import FatalError, StopProcessing +from ingestify.main import get_dev_engine +from ingestify.utils import utcnow + + +def good_loader(file_resource, current_file, **kwargs): + return DraftFile.from_input("data", data_feed_key="f1") + + +class SourceFatalInFindDatasets(Source): + """Source whose find_datasets fails fatally before yielding anything + (mirrors an account-level authorization error surfacing in discovery).""" + + provider = "test_provider" + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + raise FatalError("account deactivated") + yield # pragma: no cover - makes this a generator + + +class SourceFatalMidStream(Source): + """Source that yields 2 good datasets, then fails fatally.""" + + provider = "test_provider" + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + for i in range(2): + r = DatasetResource( + dataset_resource_id={"item_id": i}, + provider=self.provider, + dataset_type="test", + name=f"item-{i}", + ) + r.add_file( + last_modified=utcnow(), + data_feed_key="f1", + data_spec_version="v1", + file_loader=good_loader, + ) + yield r + raise FatalError("account deactivated") + + +def _setup(engine, source): + dsv = DataSpecVersionCollection.from_dict({"default": {"v1"}}) + engine.add_ingestion_plan( + IngestionPlan( + source=source, + fetch_policy=FetchPolicy(), + dataset_type="test", + selectors=[Selector.build({}, data_spec_versions=dsv)], + data_spec_versions=dsv, + ) + ) + + +def test_fatal_error_has_exit_code(): + assert FatalError.exit_code == 1 + + +def test_fatal_error_in_find_datasets_propagates(engine): + """FatalError raised in find_datasets propagates out of engine.run() + instead of being swallowed as a skipped find_datasets.""" + _setup(engine, SourceFatalInFindDatasets("s")) + + with pytest.raises(FatalError, match="deactivated"): + engine.run() + + +def test_fatal_error_mid_stream_propagates(engine): + """FatalError raised after some datasets still propagates.""" + _setup(engine, SourceFatalMidStream("s")) + + with pytest.raises(FatalError, match="deactivated"): + engine.run() + + +def test_fatal_error_persists_failed_summary(engine): + """FatalError must persist the summary (as FAILED) before aborting, so the + failure isn't lost — the loader has no finally to save it otherwise.""" + _setup(engine, SourceFatalInFindDatasets("s")) + + with patch.object(engine.store, "save_ingestion_job_summary") as mock_save: + with pytest.raises(FatalError): + engine.run() + + assert mock_save.call_count >= 1, "summary was never saved on FatalError" + saved = mock_save.call_args[0][0] + assert saved.state == IngestionJobState.FAILED + + +# --- async submit/collect path ------------------------------------------------- + + +class _AsyncSource(Source): + """Async source that raises `error` from collect() after submitting.""" + + provider = "fake_async" + + def __init__(self, name, keywords, error): + super().__init__(name) + self._keywords = keywords + self._error = error + self._in_flight = 0 + + def find_datasets( + self, dataset_type, data_spec_versions, dataset_collection_metadata, **kwargs + ): + for keyword in self._keywords: + yield DatasetResource( + dataset_resource_id={"keyword": keyword}, + provider=self.provider, + dataset_type="keyword", + name=keyword, + ) + + def submit(self, dataset_resources: Iterator[DatasetResource]) -> bool: + for _ in dataset_resources: + self._in_flight += 1 + return True + + def collect(self) -> Iterator[DatasetResource]: + raise self._error + yield # pragma: no cover - makes this a generator + + def has_pending(self) -> bool: + return self._in_flight > 0 + + +def _run_async(source, tmp_path): + engine = get_dev_engine( + source=source, + dataset_type="keyword", + data_spec_versions={"default": "v1"}, + dev_dir=str(tmp_path), + configure_logging=False, + ) + return engine + + +def test_fatal_error_in_async_collect_persists_failed_summary(tmp_path): + """A FatalError raised while collecting (the submit/collect path bigintel + uses) still persists a FAILED summary before aborting.""" + engine = _run_async(_AsyncSource("s", ["a", "b"], FatalError("boom")), tmp_path) + + with pytest.raises(FatalError): + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.FAILED + + +def test_stop_processing_in_async_collect_persists_summary(tmp_path): + """StopProcessing while collecting is a controlled stop: the summary is + persisted (FINISHED) instead of being lost, mirroring the sync path.""" + engine = _run_async( + _AsyncSource("s", ["a", "b"], StopProcessing("quota")), tmp_path + ) + + with pytest.raises(StopProcessing): + engine.run() + + summaries = engine.store.dataset_repository.load_ingestion_job_summaries() + assert len(summaries) == 1 + assert summaries[0].state == IngestionJobState.FINISHED