From 5824b8f06e9642cbf0ece29869482e774f077c72 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Wed, 5 Aug 2026 15:14:36 +0200 Subject: [PATCH] Wait for the processing of start items --- scrapy/core/engine.py | 4 +--- scrapy/core/scraper.py | 21 +++++++++++++++++++-- tests/test_spider_start.py | 28 ++++++++++++++++++++++++++-- 3 files changed, 46 insertions(+), 7 deletions(-) diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index efac5f71c..3829fd769 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -291,9 +291,7 @@ class ExecutionEngine: self.crawl(item_or_request) else: assert self._slot is not None - _schedule_coro( - self.scraper.start_itemproc_async(item_or_request, response=None) - ) + self.scraper._start_itemproc_nowait(item_or_request) self._slot.nextcall.schedule() async def _start_request_processing(self) -> None: diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index 58e37ce5e..aabd68602 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -65,7 +65,7 @@ class Slot: self.queue: deque[QueueTuple] = deque() self.active: set[Request] = set() self.active_size: int = 0 - self.itemproc_size: int = 0 # just for scrapy.utils.engine.get_engine_status() + self.itemproc_size: int = 0 self.closing: Deferred[Spider] | None = None def add_response_request( @@ -93,7 +93,7 @@ class Slot: self.active_size -= self.MIN_RESPONSE_SIZE def is_idle(self) -> bool: - return not (self.queue or self.active) + return not (self.queue or self.active or self.itemproc_size) def needs_backout(self) -> bool: return self.active_size > self.max_active_size @@ -477,6 +477,23 @@ class Scraper: ) return deferred_from_coro(self.start_itemproc_async(item, response=response)) + def _start_itemproc_nowait(self, item: Any) -> None: + """Send *item* to the item pipelines without waiting for the outcome. + + The scraper slot counts the item as being processed straight away, so + that the spider is not considered idle before the item pipelines have + had a chance to run. + """ + assert self.slot is not None # typing + self.slot.itemproc_size += 1 + _schedule_coro(self._start_itemproc_nowait_async(item)) + + async def _start_itemproc_nowait_async(self, item: Any) -> None: + assert self.slot is not None # typing + # start_itemproc_async() takes over the count of this item. + self.slot.itemproc_size -= 1 + await self.start_itemproc_async(item, response=None) + async def start_itemproc_async( self, item: Any, *, response: Response | Failure | None ) -> None: diff --git a/tests/test_spider_start.py b/tests/test_spider_start.py index 7cef2fefa..ffd39e53c 100644 --- a/tests/test_spider_start.py +++ b/tests/test_spider_start.py @@ -23,7 +23,10 @@ ITEM_B = {"id": "b"} class TestMain: async def _test_spider( - self, spider: type[Spider], expected_items: list[Any] | None = None + self, + spider: type[Spider], + expected_items: list[Any] | None = None, + settings: dict[str, Any] | None = None, ) -> None: actual_items = [] expected_items = [] if expected_items is None else expected_items @@ -31,7 +34,7 @@ class TestMain: def track_item(item, response, spider): actual_items.append(item) - crawler = get_crawler(spider) + crawler = get_crawler(spider, settings) crawler.signals.connect(track_item, signals.item_scraped) await crawler.crawl_async() assert crawler.stats @@ -98,3 +101,24 @@ class TestMain: yield ITEM_A await self._test_start(start, [ITEM_A]) + + @pytest.mark.requires_reactor # needs a reactor for twisted_sleep() + @coroutine_test + async def test_slow_pipeline(self): + class SlowPipeline: + async def process_item(self, item): + await maybe_deferred_to_future(twisted_sleep(SLEEP_SECONDS)) + return item + + class TestSpider(Spider): + name = "test" + + async def start(self): + yield ITEM_A + yield ITEM_B + + await self._test_spider( + TestSpider, + [ITEM_A, ITEM_B], + {"ITEM_PIPELINES": {SlowPipeline: 0}}, + )