Replace BatchLoader with submit/collect async source pattern

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
 

ingestify/__init__.py

@@ -7,9 +7,8 @@
 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
 
-__version__ = "0.16.0"
+__version__ = "0.17.0"