From 405e518d045587c6cf7dca07d39db0200589c993 Mon Sep 17 00:00:00 2001 From: danialza Date: Sun, 5 Jul 2026 02:39:11 +0100 Subject: [PATCH] wait for start item pipelines before closing --- scrapy/core/scraper.py | 3 +- tests/test_core_scraper.py | 58 +++++++++++++++++++++++++++++++++++++- 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index 58e37ce5e..ac36c9ee0 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -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 @@ -540,3 +540,4 @@ class Scraper: ) finally: self.slot.itemproc_size -= 1 + self._check_if_closing() diff --git a/tests/test_core_scraper.py b/tests/test_core_scraper.py index 1b8ba08ab..527084ea7 100644 --- a/tests/test_core_scraper.py +++ b/tests/test_core_scraper.py @@ -1,7 +1,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING +import asyncio +from typing import TYPE_CHECKING, Any, ClassVar +from scrapy import Spider +from scrapy.core.scraper import Scraper from scrapy.utils.test import get_crawler from tests.spiders import SimpleSpider from tests.utils.decorators import coroutine_test @@ -25,3 +28,56 @@ async def test_scraper_exception( ) await crawler.crawl_async(url=mockserver.url("/")) assert "Scraper bug processing" in caplog.text + + +@coroutine_test +async def test_close_spider_async_waits_for_item_processing() -> None: + process_started = asyncio.Event() + finish_processing = asyncio.Event() + + class BlockingPipeline: + events: ClassVar[list[str]] = [] + + async def process_item(self, item: Any) -> Any: + self.events.append("process_item_start") + process_started.set() + await finish_processing.wait() + self.events.append("process_item_end") + return item + + async def close_spider(self) -> None: + self.events.append("close_spider") + + class TestSpider(Spider): + name = "test" + + crawler = get_crawler( + TestSpider, + { + "ITEM_PIPELINES": { + BlockingPipeline: 0, + }, + }, + ) + crawler.spider = crawler._create_spider() + scraper = Scraper(crawler) + await scraper.open_spider_async() + + itemproc_task = asyncio.create_task(scraper.start_itemproc_async({}, response=None)) + await process_started.wait() + + close_task = asyncio.create_task(scraper.close_spider_async()) + try: + await asyncio.sleep(0) + assert not close_task.done() + assert BlockingPipeline.events == ["process_item_start"] + finally: + finish_processing.set() + await itemproc_task + await close_task + + assert BlockingPipeline.events == [ + "process_item_start", + "process_item_end", + "close_spider", + ]