From 12dff513d7600759d52068d5a8e04885f2ca8067 Mon Sep 17 00:00:00 2001 From: mrfandu1 <1nt22cs054.chandan@nmit.ac.in> Date: Mon, 13 Jul 2026 12:20:22 +0545 Subject: [PATCH] Store full feed batches before spider closes (#7730) Start closing and storing each batch as soon as it reaches the configured item count. Track unfinished close tasks so spider shutdown still waits for all deliveries before emitting the exporter-closed signal. Add an end-to-end regression test that verifies the first batch is stored while the crawl is still running. --- scrapy/extensions/feedexport.py | 39 +++++++++++++++++++++++-------- tests/test_feedexport_batch.py | 41 +++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index fa708a2e1..a3b35b3ac 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -13,7 +13,7 @@ import re import sys import warnings from abc import ABC, abstractmethod -from collections.abc import Callable, Coroutine +from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path, PureWindowsPath from tempfile import NamedTemporaryFile @@ -458,7 +458,7 @@ class FeedExporter: self.feeds = {} self.slots: list[FeedSlot] = [] self.filters: dict[str, ItemFilter] = {} - self._pending_close_coros: list[Coroutine[Any, Any, None]] = [] + self._pending_close_tasks: list[asyncio.Task[None] | Deferred[None]] = [] if not self.settings["FEEDS"] and not self.settings["FEED_URI"]: raise NotConfigured @@ -522,23 +522,44 @@ class FeedExporter: ) async def close_spider(self, spider: Spider) -> None: - self._pending_close_coros.extend( - self._close_slot(slot, spider) for slot in self.slots - ) + for slot in self.slots: + self._schedule_slot_close(slot, spider) - if self._pending_close_coros: + if self._pending_close_tasks: if is_asyncio_available(): await asyncio.wait( - [asyncio.create_task(coro) for coro in self._pending_close_coros] + cast("list[asyncio.Task[None]]", list(self._pending_close_tasks)) ) else: await DeferredList( - deferred_from_coro(coro) for coro in self._pending_close_coros + cast("list[Deferred[None]]", list(self._pending_close_tasks)) ) # Send FEED_EXPORTER_CLOSED signal await self.crawler.signals.send_catch_log_async(signals.feed_exporter_closed) + def _schedule_slot_close( + self, slot: FeedSlot, spider: Spider + ) -> asyncio.Task[None] | Deferred[None]: + """Start closing the slot without waiting for it to finish, keeping + track of the pending work so that it can be awaited in + :meth:`close_spider` if it hasn't finished by then.""" + aw: asyncio.Task[None] | Deferred[None] + coro = self._close_slot(slot, spider) + if is_asyncio_available(): + aw = asyncio.create_task(coro) + self._pending_close_tasks.append(aw) + aw.add_done_callback(self._pending_close_tasks.remove) + else: + aw = deferred_from_coro(coro) + self._pending_close_tasks.append(aw) + aw.addBoth(self._untrack_pending_close_task, aw) + return aw + + def _untrack_pending_close_task(self, result: Any, aw: Deferred[None]) -> Any: + self._pending_close_tasks.remove(aw) + return result + @staticmethod def _get_file(slot_: FeedSlot) -> IO[bytes]: assert slot_.file @@ -635,7 +656,7 @@ class FeedExporter: uri_params = self._get_uri_params( spider, self.feeds[slot.uri_template]["uri_params"], slot ) - self._pending_close_coros.append(self._close_slot(slot, spider)) + self._schedule_slot_close(slot, spider) slots.append( self._start_new_batch( batch_id=slot.batch_id + 1, diff --git a/tests/test_feedexport_batch.py b/tests/test_feedexport_batch.py index 469985288..8ba09a090 100644 --- a/tests/test_feedexport_batch.py +++ b/tests/test_feedexport_batch.py @@ -210,6 +210,47 @@ class TestBatchDeliveries(TestFeedExportBase): header = self.MyItem.fields.keys() await self.assertExported(items, header, rows, settings=settings) + @coroutine_test + async def test_batch_delivered_when_full(self): + """Full batches must be finalized and delivered as soon as they are + full, instead of when the spider closes (#7730).""" + dir_path = self._random_temp_filename() + batch1_path = Path(dir_path, "1.json") + mockserver_url = self.mockserver.url("/") + batch1_contents: list[bytes | None] = [] + + class TestSpider(scrapy.Spider): + name = "testspider" + start_urls = [mockserver_url] + + def parse(self, response): + yield {"foo": "bar1"} + yield {"foo": "bar2"} + yield scrapy.Request( + mockserver_url, callback=self.parse2, dont_filter=True + ) + + def parse2(self, response): + # the first batch was full after the second item, so it must + # have been delivered by now + batch1_contents.append( + batch1_path.read_bytes() if batch1_path.exists() else None + ) + yield {"foo": "bar3"} + + settings = { + "FEEDS": { + build_url(dir_path / "%(batch_id)d.json"): {"format": "json"}, + }, + "FEED_EXPORT_BATCH_ITEM_COUNT": 2, + } + crawler = get_crawler(TestSpider, settings) + await crawler.crawl_async() + + assert batch1_contents, "the second request was not processed" + assert batch1_contents[0] is not None, "batch 1 was not stored during the crawl" + assert json.loads(batch1_contents[0]) == [{"foo": "bar1"}, {"foo": "bar2"}] + def test_wrong_path(self): """If path is without %(batch_time)s and %(batch_id) an exception must be raised""" settings = {