freedak/Scrapling
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(spiders): Make graceful shutdown always enabled

Browse Source
This commit is contained in:
Karim shoair
2026-01-19 16:40:02 +02:00
parent 94bc759c6a
commit 6947066761
2 changed files with 28 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scrapling/spiders/engine.py
+18 -18
View File
@@ -230,29 +230,29 @@ class CrawlerEngine:
# Process queue # Process queue
async with create_task_group() as tg: async with create_task_group() as tg:
while self._running: while self._running:
# Check for pause/stop request if self._pause_requested:
if self._checkpoint_system_enabled: if self._active_tasks == 0 or self._force_stop:
if self._pause_requested: if self._force_stop:
# Wait for active tasks to complete log.warning(f"Force stopping with {self._active_tasks} active tasks")
if self._active_tasks == 0 or self._force_stop: tg.cancel_scope.cancel()
if self._force_stop:
log.warning(f"Force stopping with {self._active_tasks} active tasks")
# Only save checkpoint if checkpoint system is enabled
if self._checkpoint_system_enabled:
await self._save_checkpoint() await self._save_checkpoint()
self.paused = True self.paused = True
self._running = False log.info("Spider paused, checkpoint saved")
else:
log.info("Spider stopped gracefully")
if not self._force_stop: self._running = False
log.info("Spider paused, checkpoint saved") break
else:
tg.cancel_scope.cancel()
break
# Wait briefly and check again
await anyio.sleep(0.05)
continue
if self._is_checkpoint_time(): # Wait briefly and check again
await self._save_checkpoint() await anyio.sleep(0.05)
continue
if self._checkpoint_system_enabled and self._is_checkpoint_time():
await self._save_checkpoint()
if self.scheduler.is_empty: if self.scheduler.is_empty:
# Empty queue + no active tasks = done # Empty queue + no active tasks = done
scrapling/spiders/spider.py
+10 -12
View File
@@ -211,13 +211,11 @@ class Spider(ABC):
manager.add("default", FetcherSession()) manager.add("default", FetcherSession())
def pause(self): def pause(self):
"""Pause the crawling process. Requires crawldir to be set for checkpoint system.""" """Request graceful shutdown of the crawling process."""
if not self.crawldir:
raise RuntimeError("Cannot pause without crawldir - checkpoint system not enabled")
if self._engine: if self._engine:
self._engine.request_pause() self._engine.request_pause()
else: else:
raise RuntimeError("Spider doesn't have active crawl to pause, no crawl engine started!") raise RuntimeError("No active crawl to stop")
def _setup_signal_handler(self) -> None: def _setup_signal_handler(self) -> None:
"""Set up SIGINT handler for graceful pause.""" """Set up SIGINT handler for graceful pause."""
@@ -264,8 +262,11 @@ class Spider(ABC):
This is the main entry point for running a spider. This is the main entry point for running a spider.
Handles async execution internally via anyio. Handles async execution internally via anyio.
If crawldir is set, pressing Ctrl+C will pause the spider and save a checkpoint. Pressing Ctrl+C will initiate graceful shutdown (waits for active tasks to complete).
Running the spider again with the same crawldir will resume from the checkpoint. Pressing Ctrl+C a second time will force immediate stop.
If crawldir is set, a checkpoint will also be saved on graceful shutdown,
allowing you to resume the crawl later by running the spider again.
:param use_uvloop: Whether to use the faster uvloop/winloop event loop implementation, if available. :param use_uvloop: Whether to use the faster uvloop/winloop event loop implementation, if available.
:param backend_options: Asyncio backend options to be used with `anyio.run` :param backend_options: Asyncio backend options to be used with `anyio.run`
@@ -274,15 +275,12 @@ class Spider(ABC):
if use_uvloop: if use_uvloop:
backend_options.update({"use_uvloop": True}) backend_options.update({"use_uvloop": True})
# Set up SIGINT handler for graceful pause (only if crawldir is set) # Set up SIGINT handler for graceful shutdown
if self.crawldir: self._setup_signal_handler()
self._setup_signal_handler()
try: try:
return anyio.run(self.__run, backend="asyncio", backend_options=backend_options) return anyio.run(self.__run, backend="asyncio", backend_options=backend_options)
finally: finally:
if self.crawldir: self._restore_signal_handler()
self._restore_signal_handler()
async def stream(self) -> AsyncGenerator[Dict[str, Any], None]: async def stream(self) -> AsyncGenerator[Dict[str, Any], None]:
"""Stream items as they're scraped. Ideal for long-running spiders or building applications on top of the spiders. """Stream items as they're scraped. Ideal for long-running spiders or building applications on top of the spiders.