From 5ddde1fdbb115bae72994a7ddbe0de4de4f01e58 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Wed, 20 May 2026 15:59:31 +0200 Subject: [PATCH 1/4] Replace BatchLoader with submit/collect async source pattern MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sources can now implement submit(), collect(), and has_pending() for async bulk fetching. Ingestify detects these via duck typing and uses a streaming pipeline: submit until capacity full, collect results as they arrive, submit more — interleaved with lazy find_datasets. - Add _execute_async path in IngestionJob for sources with submit/collect - Add DatasetResource.update_file() for setting file content by data_feed_key - Add test_async_source.py with 4 tests (basic, capacity, skip, mixed files) - Remove BatchLoader, BatchTask, _wrap_batch_tasks (replaced by submit/collect) - Update docs/api_reference.md --- docs/api_reference.md | 83 ++---- ingestify/__init__.py | 1 - .../domain/models/ingestion/ingestion_job.py | 272 +++++++++++------- .../domain/models/resources/batch_loader.py | 51 ---- .../models/resources/dataset_resource.py | 9 + ingestify/tests/test_async_source.py | 179 ++++++++++++ ingestify/tests/test_batch_loader.py | 39 --- 7 files changed, 381 insertions(+), 253 deletions(-) delete mode 100644 ingestify/domain/models/resources/batch_loader.py create mode 100644 ingestify/tests/test_async_source.py delete mode 100644 ingestify/tests/test_batch_loader.py diff --git a/docs/api_reference.md b/docs/api_reference.md index 25becec..95e0fe6 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -216,65 +216,44 @@ class CustomSource(Source): yield dataset_resource ``` -### BatchLoader (batching file loads) +### Async Source (submit/collect) -When the underlying data source is more efficient with batched requests (e.g. an API that accepts many items per call), wrap your loader function in a `BatchLoader` and share the instance across the `DatasetResource`s that should be batched together. - -Ingestify groups resources by shared `BatchLoader` instance, chunks them into groups of `batch_size`, and calls the wrapped loader once per chunk. +For APIs where you submit work and collect results later (e.g. async SERP APIs, +batch processing endpoints), a source can implement `submit()`, `collect()`, and +`has_pending()`. Ingestify detects these automatically — no special base class needed. ```python -from ingestify import BatchLoader, DatasetResource, Source -from ingestify.domain import DraftFile - - -def load_metrics(file_resources, current_files, dataset_resources): - """Called once per batch. Each argument is a list of up to batch_size items. - - current_files may contain None (for new datasets) or a File (for existing - ones). Return a list of DraftFile / NotModifiedFile in the same order. - """ - identifiers = [dr.dataset_resource_id for dr in dataset_resources] - results = fetch_batch_from_api(identifiers) # single API call - return [ - DraftFile.from_input( - file_=json.dumps(result), - data_serialization_format="json", - data_feed_key=fr.data_feed_key, - data_spec_version=fr.data_spec_version, - modified_at=fr.last_modified, - ) - for fr, result in zip(file_resources, results) - ] - - class MySource(Source): - provider = "my_provider" - - def find_datasets(self, dataset_type, data_spec_versions, **selector): - # Share one BatchLoader instance across all DatasetResources that - # should be batched together. - batch_loader = BatchLoader(load_metrics, batch_size=20) - - for item_id in items: - resource = DatasetResource( - dataset_resource_id={"item_id": item_id}, - dataset_type=dataset_type, - provider=self.provider, - name=str(item_id), - ) - resource.add_file( - last_modified=last_modified, - data_feed_key="metrics", - data_spec_version="v1", - file_loader=batch_loader, - data_serialization_format="json", - ) + provider = "my_api" + + def find_datasets(self, dataset_type, data_spec_versions, **kwargs): + for item in discover_items(): + yield DatasetResource( + dataset_resource_id={"item_id": item.id}, ... + ).add_file(last_modified=item.updated_at, data_feed_key="data", data_spec_version="v1") + + def submit(self, dataset_resources): + """Pull from iterator until capacity full. Return True when exhausted.""" + if self._in_flight >= MAX: + return False + for resource in dataset_resources: + self._send_to_api(resource) + if self._in_flight >= MAX: + return False + return True + + def collect(self): + """Yield resources with file content set as results arrive.""" + for resource, result in self._poll_completed(): + resource.update_file("data", json_content=result) yield resource + + def has_pending(self): + return self._in_flight > 0 ``` -Notes: -- Only resources that actually need loading (i.e. not skipped by `FetchPolicy`) are passed to the loader. -- The last batch of a group may contain fewer than `batch_size` items if the number of pending items is not a multiple of `batch_size`, or if the group crosses an internal chunk boundary. +Ingestify filters before `submit` — only resources that need fetching are passed in. +The loop alternates: submit until full → collect what's ready → submit more. ### Custom Event Subscriber diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 0ba8c7a..8c3678d 100644 --- a/ingestify/__init__.py +++ b/ingestify/__init__.py @@ -7,7 +7,6 @@ if not __INGESTIFY_SETUP__: from .infra import retrieve_http from .source_base import Source, DatasetResource - from .domain.models.resources.batch_loader import BatchLoader from .exceptions import StopProcessing from .main import debug_source diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 7704eaa..1e61d65 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -23,7 +23,6 @@ FileResource, DatasetResource, ) -from ingestify.domain.models.resources.batch_loader import BatchLoader from ingestify.domain.models.dataset.dataset import DatasetLastModifiedAtMap from ingestify.domain.models.task.task_summary import TaskSummary from ingestify.exceptions import SaveError, IngestifyError, StopProcessing @@ -118,12 +117,7 @@ def load_file( @lru_cache(maxsize=None) def _loader_accepts_dataset_resource(loader) -> bool: - """Return True if loader accepts a `dataset_resource` keyword argument. - - BatchLoader instances always do. Plain functions are introspected once. - """ - if isinstance(loader, BatchLoader): - return True + """Return True if loader accepts a `dataset_resource` keyword argument.""" try: sig = inspect.signature(loader) except (TypeError, ValueError): @@ -242,96 +236,6 @@ def __repr__(self): return f"CreateDatasetTask({self.dataset_resource.provider} -> {self.dataset_resource.dataset_resource_id})" -class BatchTask(Task): - """Wraps a group of inner tasks that share a BatchLoader instance. - - On run(), invokes the shared loader_fn once with all items in the batch, - caches the results, then runs each inner task sequentially. - """ - - def __init__(self, inner_tasks: list, loader: BatchLoader): - self.inner_tasks = inner_tasks - self.loader = loader - - def run(self): - # Collect items for the shared loader across all inner tasks. - file_resources, current_files, dataset_resources = [], [], [] - for task in self.inner_tasks: - dataset = getattr(task, "dataset", None) - for file_id, file_resource in task.dataset_resource.files.items(): - # A DatasetResource can have multiple files, each potentially - # using a different file_loader (e.g. a plain loader for one - # file and a BatchLoader for another, or multiple BatchLoaders). - # We only want files whose loader is this BatchTask's loader. - if file_resource.file_loader is not self.loader: - continue - current_file = None - if dataset is not None: - current_file = dataset.current_revision.modified_files_map.get( - file_id - ) - file_resources.append(file_resource) - current_files.append(current_file) - dataset_resources.append(task.dataset_resource) - - results = self.loader.loader_fn( - file_resources, current_files, dataset_resources - ) - if len(results) != len(file_resources): - raise RuntimeError( - f"BatchLoader expected {len(file_resources)} results, got {len(results)}" - ) - self.loader._store_results(file_resources, results) - - # Run the wrapped inner tasks — they pick up cached results via - # BatchLoader.__call__ inside load_file(). - return [task.run() for task in self.inner_tasks] - - def __repr__(self): - return f"BatchTask(n={len(self.inner_tasks)})" - - -def _wrap_batch_tasks(task_set: "TaskSet") -> "TaskSet": - """Rebuild a TaskSet, wrapping tasks that share a BatchLoader instance - into BatchTasks chunked by the loader's batch_size. - - Tasks with no BatchLoader in any of their files remain unchanged. - """ - loose: list = [] - grouped: dict = {} # id(loader) -> (loader, [tasks]) - - for task in task_set: - loader = _find_first_batch_loader(task) - if loader is None: - loose.append(task) - else: - grouped.setdefault(id(loader), (loader, []))[1].append(task) - - new_task_set = TaskSet() - for task in loose: - new_task_set.add(task) - - for loader, tasks in grouped.values(): - batch_size = loader.batch_size - for i in range(0, len(tasks), batch_size): - new_task_set.add( - BatchTask(inner_tasks=tasks[i : i + batch_size], loader=loader) - ) - - return new_task_set - - -def _find_first_batch_loader(task) -> "Optional[BatchLoader]": - """Return the first BatchLoader encountered in the task's file resources.""" - dataset_resource = getattr(task, "dataset_resource", None) - if dataset_resource is None: - return None - for file_resource in dataset_resource.files.values(): - if isinstance(file_resource.file_loader, BatchLoader): - return file_resource.file_loader - return None - - MAX_TASKS_PER_CHUNK = 10_000 @@ -417,6 +321,14 @@ def execute( logger.info("Starting tasks") + source = self.ingestion_plan.source + if hasattr(source, "submit") and hasattr(source, "collect"): + yield from self._execute_async( + source, batches, store, task_executor, + last_modified_at_map, ingestion_job_summary, is_first_chunk, + ) + return + while True: logger.info(f"Finding next batch of datasets for selector={self.selector}") @@ -521,13 +433,10 @@ def execute( with ingestion_job_summary.record_timing("tasks"): if task_set: - original_task_count = len(task_set) - task_set = _wrap_batch_tasks(task_set) logger.info( f"Discovered {len(dataset_identifiers)} datasets from {self.ingestion_plan.source.__class__.__name__} " - f"using selector {self.selector} => {original_task_count} tasks. {skipped_tasks} skipped." + f"using selector {self.selector} => {len(task_set)} tasks. {skipped_tasks} skipped." ) - logger.info(f"Running {len(task_set)} tasks") try: results = task_executor.run(run_task, task_set) @@ -540,15 +449,7 @@ def execute( yield ingestion_job_summary raise - # BatchTasks return a list of TaskSummary; flatten. - task_summaries = [] - for result in results: - if isinstance(result, list): - task_summaries.extend(result) - else: - task_summaries.append(result) - - ingestion_job_summary.add_task_summaries(task_summaries) + ingestion_job_summary.add_task_summaries(results) else: logger.info( f"Discovered {len(dataset_identifiers)} datasets from {self.ingestion_plan.source.__class__.__name__} " @@ -568,3 +469,154 @@ def execute( # When there is interesting information to store, or there was no data at all, store it ingestion_job_summary.set_finished() yield ingestion_job_summary + + def _execute_async( + self, + source, + batches, + store: DatasetStore, + task_executor: TaskExecutor, + last_modified_at_map, + ingestion_job_summary: IngestionJobSummary, + is_first_chunk: bool, + ) -> Iterator[IngestionJobSummary]: + """Execute using the submit/collect pattern for async sources.""" + + def filtered_stream(): + """Lazily yield filtered DatasetResources across all batches.""" + while True: + try: + with ingestion_job_summary.record_timing("find_datasets"): + try: + batch = next(batches) + except StopIteration: + return + except Exception as e: + logger.exception("Failed to fetch next batch") + ingestion_job_summary.set_exception(e) + return + + # Fast pre-check + if last_modified_at_map: + pending = [] + for dr in batch: + identifier = Identifier.create_from_selector( + self.selector, **dr.dataset_resource_id + ) + ts = last_modified_at_map.get(identifier.key) + if ts is not None and dr.files: + max_mod = max(f.last_modified for f in dr.files.values()) + if ts >= max_mod: + ingestion_job_summary.increase_skipped_tasks(1) + continue + pending.append(dr) + batch = pending + + if not batch: + continue + + # Store check: determine create vs update + dataset_identifiers = [ + Identifier.create_from_selector( + self.selector, **dr.dataset_resource_id + ) + for dr in batch + ] + + with ingestion_job_summary.record_timing("get_dataset_collection"): + dataset_collection = store.get_dataset_collection( + dataset_type=self.ingestion_plan.dataset_type, + provider=batch[0].provider, + selector=dataset_identifiers, + ) + + for dr in batch: + identifier = Identifier.create_from_selector( + self.selector, **dr.dataset_resource_id + ) + dataset = dataset_collection.get(identifier) + if dataset: + if self.ingestion_plan.fetch_policy.should_refetch(dataset, dr): + dr._existing_dataset = dataset + yield dr + else: + store.dispatch(DatasetSkipped(dataset=dataset)) + ingestion_job_summary.increase_skipped_tasks(1) + else: + if self.ingestion_plan.fetch_policy.should_fetch(dr): + yield dr + else: + ingestion_job_summary.increase_skipped_tasks(1) + + 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]) + + if ingestion_job_summary.task_count() > 0 or is_first_chunk: + ingestion_job_summary.set_finished() + yield ingestion_job_summary + + def _store_async_result(self, dataset_resource: DatasetResource, store: DatasetStore): + """Store a dataset resource returned by collect().""" + import uuid + + dataset_identifier = Identifier(**dataset_resource.dataset_resource_id) + revision_source = RevisionSource( + source_id=str(uuid.uuid1()), source_type=SourceType.TASK + ) + + existing_dataset = getattr(dataset_resource, "_existing_dataset", None) + + # Load files that have file_loader or json_content + files = {} + for file_id, file_resource in dataset_resource.files.items(): + files[file_id] = load_file( + file_resource, + dataset=existing_dataset, + dataset_resource=dataset_resource, + ) + + if existing_dataset: + with TaskSummary.update( + str(uuid.uuid1()), dataset_identifier=dataset_identifier + ) as task_summary: + dataset_resource.run_post_load_files(files, existing_dataset) + try: + revision = store.update_dataset( + dataset=existing_dataset, + name=dataset_resource.name, + state=dataset_resource.state, + metadata=dataset_resource.metadata, + files=files, + revision_source=revision_source, + ) + task_summary.set_stats_from_revision(revision) + except Exception as e: + raise SaveError("Could not update dataset") from e + else: + with TaskSummary.create( + str(uuid.uuid1()), dataset_identifier + ) as task_summary: + dataset_resource.run_post_load_files(files) + try: + revision = store.create_dataset( + dataset_type=dataset_resource.dataset_type, + provider=dataset_resource.provider, + dataset_identifier=dataset_identifier, + name=dataset_resource.name, + state=dataset_resource.state, + metadata=dataset_resource.metadata, + files=files, + revision_source=revision_source, + ) + task_summary.set_stats_from_revision(revision) + except Exception as e: + raise SaveError("Could not create dataset") from e + + return task_summary diff --git a/ingestify/domain/models/resources/batch_loader.py b/ingestify/domain/models/resources/batch_loader.py deleted file mode 100644 index fa8f20f..0000000 --- a/ingestify/domain/models/resources/batch_loader.py +++ /dev/null @@ -1,51 +0,0 @@ -"""BatchLoader wraps a file loader so multiple files are fetched in one call. - -Use this when the data source is more efficient with batched requests. Create -one BatchLoader instance and share it across the DatasetResources that should -be batched together. Ingestify groups those resources by shared BatchLoader -instance, chunks them into groups of `batch_size`, and calls the wrapped -loader_fn once per chunk. - -The wrapped loader_fn receives lists instead of single items and must return -a list of results in the same order: - - def load(file_resources, current_files, dataset_resources): - ... - return [DraftFile.from_input(...) for _ in file_resources] - - batch_loader = BatchLoader(load, batch_size=20) - resource.add_file(file_loader=batch_loader, ...) - -current_files may contain None entries (for create tasks) or a File (for -update tasks) — the loader_fn handles both. - -Error propagation: if the loader_fn stores an Exception instead of a -DraftFile for a specific item, __call__ will re-raise it when that item's -task runs. This lets successful items in a batch proceed while failed items -cause their individual tasks to fail and be retried on the next run. -""" -from typing import Callable, List - - -class BatchLoader: - def __init__(self, loader_fn: Callable, batch_size: int): - self.loader_fn = loader_fn - self.batch_size = batch_size - self._results: dict = {} - - def __call__(self, file_resource, current_file, dataset_resource=None, **kwargs): - key = id(file_resource) - if key not in self._results: - raise RuntimeError( - "BatchLoader result not precomputed. A BatchTask must " - "populate the cache before inner tasks execute." - ) - result = self._results.pop(key) - if isinstance(result, BaseException): - raise result - return result - - def _store_results(self, file_resources: List, results: List): - """Store batch results so they can be retrieved via __call__.""" - for file_resource, result in zip(file_resources, results): - self._results[id(file_resource)] = result diff --git a/ingestify/domain/models/resources/dataset_resource.py b/ingestify/domain/models/resources/dataset_resource.py index 293073e..250c9c3 100644 --- a/ingestify/domain/models/resources/dataset_resource.py +++ b/ingestify/domain/models/resources/dataset_resource.py @@ -106,3 +106,12 @@ def add_file( # Allow chaining return self + + def update_file(self, data_feed_key: str, **kwargs): + """Update an existing file resource's attributes (e.g. json_content).""" + for file_id, file_resource in self.files.items(): + if file_resource.data_feed_key == data_feed_key: + for key, value in kwargs.items(): + setattr(file_resource, key, value) + return self + raise KeyError(f"No file with data_feed_key '{data_feed_key}'") diff --git a/ingestify/tests/test_async_source.py b/ingestify/tests/test_async_source.py new file mode 100644 index 0000000..4cf38c5 --- /dev/null +++ b/ingestify/tests/test_async_source.py @@ -0,0 +1,179 @@ +"""Tests for sources with submit/collect (async loading pattern).""" +from typing import Iterator + +from ingestify import Source, DatasetResource +from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector +from ingestify.domain.models.dataset.collection_metadata import DatasetCollectionMetadata +from ingestify.domain.models.fetch_policy import FetchPolicy +from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan +from ingestify.main import get_dev_engine +from ingestify.utils import utcnow + + +class FakeAsyncSource(Source): + """Source that uses submit/collect instead of file_loader.""" + + provider = "fake_async" + + def __init__(self, name, datasets, capacity=3): + super().__init__(name) + self._datasets = datasets + self._capacity = capacity + self._buffer = [] + self._submitted = {} + self._in_flight = 0 + + def find_datasets( + self, + dataset_type, + data_spec_versions, + dataset_collection_metadata, + **kwargs, + ): + last_modified = utcnow() + for keyword in self._datasets: + yield DatasetResource( + dataset_resource_id={"keyword": keyword}, + provider=self.provider, + dataset_type="keyword", + name=keyword, + ) + + def submit(self, dataset_resources: Iterator[DatasetResource]) -> bool: + for resource in dataset_resources: + keyword = resource.dataset_resource_id["keyword"] + self._submitted[keyword] = resource + self._in_flight += 1 + if self._in_flight >= self._capacity: + return False + return True + + def collect(self) -> Iterator[DatasetResource]: + for keyword, resource in list(self._submitted.items()): + resource.add_file( + last_modified=utcnow(), + data_feed_key="data", + data_spec_version="v1", + json_content={"keyword": keyword, "rank": 1}, + ) + del self._submitted[keyword] + self._in_flight -= 1 + yield resource + + def has_pending(self) -> bool: + return self._in_flight > 0 + + +def _make_engine(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_async_source_basic(tmp_path): + """Source with submit/collect ingests all datasets with files.""" + source = FakeAsyncSource("test", ["alpha", "beta", "gamma"]) + engine = _make_engine(source, tmp_path) + engine.run() + + datasets = list(engine.store.get_dataset_collection(dataset_type="keyword")) + assert len(datasets) == 3 + keywords = {d.identifier["keyword"] for d in datasets} + assert keywords == {"alpha", "beta", "gamma"} + + # Verify files were actually stored + for ds in datasets: + files = engine.store.load_files(ds) + loaded = files.get_file("data") + assert loaded is not None, f"Dataset {ds.identifier} has no 'data' file" + + +def test_async_source_with_capacity(tmp_path): + """submit returns False when capacity is reached, collect frees capacity.""" + source = FakeAsyncSource("test", ["a", "b", "c", "d", "e"], capacity=2) + engine = _make_engine(source, tmp_path) + engine.run() + + datasets = list(engine.store.get_dataset_collection(dataset_type="keyword")) + assert len(datasets) == 5 + for ds in datasets: + files = engine.store.load_files(ds) + assert files.get_file("data") is not None + + +def test_async_source_skips_existing(tmp_path): + """Second run skips already-ingested datasets.""" + source = FakeAsyncSource("test", ["alpha", "beta"]) + engine = _make_engine(source, tmp_path) + + # First run + engine.run() + datasets = list(engine.store.get_dataset_collection(dataset_type="keyword")) + assert len(datasets) == 2 + + # Second run — same data, should skip + engine.run() + datasets = list(engine.store.get_dataset_collection(dataset_type="keyword")) + assert len(datasets) == 2 + # Each dataset should have exactly 1 revision (not re-fetched) + for ds in datasets: + assert len(ds.revisions) == 1 + + +def test_async_source_with_existing_files(tmp_path): + """find_datasets can attach files; collect adds more.""" + + class SourceWithExistingFiles(Source): + provider = "fake_async" + + def find_datasets(self, dataset_type, data_spec_versions, + dataset_collection_metadata, **kwargs): + yield ( + DatasetResource( + dataset_resource_id={"keyword": "test"}, + provider=self.provider, + dataset_type="keyword", + name="test", + ).add_file( + last_modified=utcnow(), + data_feed_key="metadata", + data_spec_version="v1", + json_content={"source": "find_datasets"}, + ) + ) + + def submit(self, dataset_resources): + self._resources = [] + for r in dataset_resources: + self._resources.append(r) + return True + + def collect(self): + for resource in self._resources: + resource.add_file( + last_modified=utcnow(), + data_feed_key="serp", + data_spec_version="v1", + json_content={"source": "collect", "rank": 1}, + ) + yield resource + self._resources = [] + + def has_pending(self): + return bool(self._resources) + + source = SourceWithExistingFiles("test") + engine = _make_engine(source, tmp_path) + engine.run() + + datasets = list(engine.store.get_dataset_collection(dataset_type="keyword")) + assert len(datasets) == 1 + + files = engine.store.load_files(datasets[0]) + assert files.get_file("metadata") is not None + assert files.get_file("serp") is not None diff --git a/ingestify/tests/test_batch_loader.py b/ingestify/tests/test_batch_loader.py deleted file mode 100644 index ef9b706..0000000 --- a/ingestify/tests/test_batch_loader.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Tests for BatchLoader.""" -import pytest - -from ingestify.domain.models.resources.batch_loader import BatchLoader - - -def test_batch_loader_returns_cached_result(): - """__call__ returns the result stored for each file_resource.""" - batch_loader = BatchLoader(lambda frs, cfs, drs: [], batch_size=5) - fr1, fr2 = object(), object() - - batch_loader._store_results([fr1, fr2], ["result_1", "result_2"]) - - assert batch_loader(fr1, current_file=None) == "result_1" - assert batch_loader(fr2, current_file=None) == "result_2" - - -def test_batch_loader_raises_if_not_precomputed(): - """__call__ raises when the cache has no entry for this file_resource.""" - batch_loader = BatchLoader(lambda frs, cfs, drs: [], batch_size=5) - - with pytest.raises(RuntimeError, match="not precomputed"): - batch_loader(object(), current_file=None) - - -def test_batch_loader_propagates_stored_exception(): - """When an Exception is stored as a result, __call__ re-raises it.""" - batch_loader = BatchLoader(lambda frs, cfs, drs: [], batch_size=5) - fr_ok, fr_err = object(), object() - - original_error = ValueError("Google Ads daily quota exhausted") - batch_loader._store_results([fr_ok, fr_err], ["good_result", original_error]) - - # Good result works normally - assert batch_loader(fr_ok, current_file=None) == "good_result" - - # Error result re-raises the original exception - with pytest.raises(ValueError, match="daily quota exhausted"): - batch_loader(fr_err, current_file=None) From 32fe50a00b61e6194a613a1e8bb86c38f33f36f2 Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Tue, 30 Jun 2026 16:57:26 +0200 Subject: [PATCH 2/4] Apply black 22.3.0 formatting (CI) --- ingestify/domain/models/ingestion/ingestion_job.py | 13 ++++++++++--- ingestify/tests/test_async_source.py | 13 ++++++++++--- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/ingestify/domain/models/ingestion/ingestion_job.py b/ingestify/domain/models/ingestion/ingestion_job.py index 1e61d65..772d382 100644 --- a/ingestify/domain/models/ingestion/ingestion_job.py +++ b/ingestify/domain/models/ingestion/ingestion_job.py @@ -324,8 +324,13 @@ def execute( source = self.ingestion_plan.source if hasattr(source, "submit") and hasattr(source, "collect"): yield from self._execute_async( - source, batches, store, task_executor, - last_modified_at_map, ingestion_job_summary, is_first_chunk, + source, + batches, + store, + task_executor, + last_modified_at_map, + ingestion_job_summary, + is_first_chunk, ) return @@ -562,7 +567,9 @@ def filtered_stream(): ingestion_job_summary.set_finished() yield ingestion_job_summary - def _store_async_result(self, dataset_resource: DatasetResource, store: DatasetStore): + def _store_async_result( + self, dataset_resource: DatasetResource, store: DatasetStore + ): """Store a dataset resource returned by collect().""" import uuid diff --git a/ingestify/tests/test_async_source.py b/ingestify/tests/test_async_source.py index 4cf38c5..586f986 100644 --- a/ingestify/tests/test_async_source.py +++ b/ingestify/tests/test_async_source.py @@ -3,7 +3,9 @@ from ingestify import Source, DatasetResource from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector -from ingestify.domain.models.dataset.collection_metadata import DatasetCollectionMetadata +from ingestify.domain.models.dataset.collection_metadata import ( + DatasetCollectionMetadata, +) from ingestify.domain.models.fetch_policy import FetchPolicy from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan from ingestify.main import get_dev_engine @@ -131,8 +133,13 @@ def test_async_source_with_existing_files(tmp_path): class SourceWithExistingFiles(Source): provider = "fake_async" - def find_datasets(self, dataset_type, data_spec_versions, - dataset_collection_metadata, **kwargs): + def find_datasets( + self, + dataset_type, + data_spec_versions, + dataset_collection_metadata, + **kwargs, + ): yield ( DatasetResource( dataset_resource_id={"keyword": "test"}, From 62b2ba324347ed2e6aafd85b1932090a7e644caa Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Tue, 30 Jun 2026 17:00:40 +0200 Subject: [PATCH 3/4] Drop BatchLoader import (replaced by submit/collect on this branch) --- ingestify/__init__.py | 1 - 1 file changed, 1 deletion(-) diff --git a/ingestify/__init__.py b/ingestify/__init__.py index a20086e..31603b4 100644 --- a/ingestify/__init__.py +++ b/ingestify/__init__.py @@ -7,7 +7,6 @@ if not __INGESTIFY_SETUP__: from .infra import retrieve_http from .source_base import Source, DatasetResource - from .domain.models.resources.batch_loader import BatchLoader from .domain.models.fetch_policy import FetchPolicy from .exceptions import StopProcessing from .main import debug_source From 37c3b338214fa5187a3bbcb8383ed52c4e9dc9db Mon Sep 17 00:00:00 2001 From: Koen Vossen Date: Tue, 30 Jun 2026 23:01:28 +0200 Subject: [PATCH 4/4] Bump version to 0.17.0 --- ingestify/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 31603b4..5c59e9e 100644 --- a/ingestify/__init__.py +++ b/ingestify/__init__.py @@ -11,4 +11,4 @@ from .exceptions import StopProcessing from .main import debug_source -__version__ = "0.16.0" +__version__ = "0.17.0"