From 8fb8d2c6b8a83181ef6564fbc9afa9cfeea2bd05 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 3 Jun 2025 23:21:33 +0500 Subject: [PATCH] Add AsyncioLoopingCall. (#6855) --- scrapy/core/downloader/__init__.py | 8 ++- scrapy/core/engine.py | 11 ++- scrapy/extensions/closespider.py | 5 +- scrapy/extensions/logstats.py | 12 ++-- scrapy/extensions/memusage.py | 16 +++-- scrapy/extensions/periodic_log.py | 14 ++-- scrapy/utils/asyncio.py | 106 ++++++++++++++++++++++++++++- tests/test_utils_asyncio.py | 48 ++++++++++++- 8 files changed, 195 insertions(+), 25 deletions(-) diff --git a/scrapy/core/downloader/__init__.py b/scrapy/core/downloader/__init__.py index 5468398aa..501c669ce 100644 --- a/scrapy/core/downloader/__init__.py +++ b/scrapy/core/downloader/__init__.py @@ -7,7 +7,6 @@ from datetime import datetime from time import time from typing import TYPE_CHECKING, Any, cast -from twisted.internet import task from twisted.internet.defer import Deferred, inlineCallbacks from scrapy import Request, Spider, signals @@ -15,6 +14,7 @@ from scrapy.core.downloader.handlers import DownloadHandlers from scrapy.core.downloader.middleware import DownloaderMiddlewareManager from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.resolver import dnscache +from scrapy.utils.asyncio import AsyncioLoopingCall, create_looping_call from scrapy.utils.defer import ( deferred_from_coro, maybe_deferred_to_future, @@ -25,6 +25,8 @@ from scrapy.utils.httpobj import urlparse_cached if TYPE_CHECKING: from collections.abc import Generator + from twisted.internet.task import LoopingCall + from scrapy.crawler import Crawler from scrapy.http import Response from scrapy.settings import BaseSettings @@ -111,7 +113,9 @@ class Downloader: self.middleware: DownloaderMiddlewareManager = ( DownloaderMiddlewareManager.from_crawler(crawler) ) - self._slot_gc_loop: task.LoopingCall = task.LoopingCall(self._slot_gc) + self._slot_gc_loop: AsyncioLoopingCall | LoopingCall = create_looping_call( + self._slot_gc + ) self._slot_gc_loop.start(60) self.per_slot_settings: dict[str, dict[str, Any]] = self.settings.getdict( "DOWNLOAD_SLOTS", {} diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index 721c81d81..d9361a674 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -13,13 +13,16 @@ from traceback import format_exc from typing import TYPE_CHECKING, Any, cast from twisted.internet.defer import Deferred, inlineCallbacks, succeed -from twisted.internet.task import LoopingCall from twisted.python.failure import Failure from scrapy import signals from scrapy.core.scraper import Scraper from scrapy.exceptions import CloseSpider, DontCloseSpider, IgnoreRequest from scrapy.http import Request, Response +from scrapy.utils.asyncio import ( + AsyncioLoopingCall, + create_looping_call, +) from scrapy.utils.defer import ( deferred_f_from_coro_f, deferred_from_coro, @@ -32,6 +35,8 @@ from scrapy.utils.reactor import CallLaterOnce if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable, Generator + from twisted.internet.task import LoopingCall + from scrapy.core.downloader import Downloader from scrapy.core.scheduler import BaseScheduler from scrapy.crawler import Crawler @@ -56,7 +61,9 @@ class _Slot: self.close_if_idle: bool = close_if_idle self.nextcall: CallLaterOnce[None] = nextcall self.scheduler: BaseScheduler = scheduler - self.heartbeat: LoopingCall = LoopingCall(nextcall.schedule) + self.heartbeat: AsyncioLoopingCall | LoopingCall = create_looping_call( + nextcall.schedule + ) def add_request(self, request: Request) -> None: self.inprogress.add(request) diff --git a/scrapy/extensions/closespider.py b/scrapy/extensions/closespider.py index dff8bc97e..a649a86e2 100644 --- a/scrapy/extensions/closespider.py +++ b/scrapy/extensions/closespider.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any from scrapy import Request, Spider, signals from scrapy.exceptions import NotConfigured +from scrapy.utils.asyncio import create_looping_call if TYPE_CHECKING: from twisted.python.failure import Failure @@ -118,9 +119,7 @@ class CloseSpider: task_no_item.stop() def spider_opened_no_item(self, spider: Spider) -> None: - from twisted.internet import task - - self.task_no_item = task.LoopingCall(self._count_items_produced, spider) + self.task_no_item = create_looping_call(self._count_items_produced, spider) self.task_no_item.start(self.timeout_no_item, now=False) logger.info( diff --git a/scrapy/extensions/logstats.py b/scrapy/extensions/logstats.py index f2e1f57b8..387cfddb3 100644 --- a/scrapy/extensions/logstats.py +++ b/scrapy/extensions/logstats.py @@ -3,12 +3,16 @@ from __future__ import annotations import logging from typing import TYPE_CHECKING -from twisted.internet import task - from scrapy import Spider, signals from scrapy.exceptions import NotConfigured +from scrapy.utils.asyncio import ( + AsyncioLoopingCall, + create_looping_call, +) if TYPE_CHECKING: + from twisted.internet.task import LoopingCall + # typing.Self requires Python 3.11 from typing_extensions import Self @@ -29,7 +33,7 @@ class LogStats: self.stats: StatsCollector = stats self.interval: float = interval self.multiplier: float = 60.0 / self.interval - self.task: task.LoopingCall | None = None + self.task: AsyncioLoopingCall | LoopingCall | None = None @classmethod def from_crawler(cls, crawler: Crawler) -> Self: @@ -46,7 +50,7 @@ class LogStats: self.pagesprev: int = 0 self.itemsprev: int = 0 - self.task = task.LoopingCall(self.log, spider) + self.task = create_looping_call(self.log, spider) self.task.start(self.interval) def log(self, spider: Spider) -> None: diff --git a/scrapy/extensions/memusage.py b/scrapy/extensions/memusage.py index d7f810107..2ef322f1a 100644 --- a/scrapy/extensions/memusage.py +++ b/scrapy/extensions/memusage.py @@ -13,14 +13,18 @@ from importlib import import_module from pprint import pformat from typing import TYPE_CHECKING -from twisted.internet import task - from scrapy import signals from scrapy.exceptions import NotConfigured from scrapy.mail import MailSender +from scrapy.utils.asyncio import ( + AsyncioLoopingCall, + create_looping_call, +) from scrapy.utils.engine import get_engine_status if TYPE_CHECKING: + from twisted.internet.task import LoopingCall + # typing.Self requires Python 3.11 from typing_extensions import Self @@ -66,16 +70,16 @@ class MemoryUsage: def engine_started(self) -> None: assert self.crawler.stats self.crawler.stats.set_value("memusage/startup", self.get_virtual_size()) - self.tasks: list[task.LoopingCall] = [] - tsk = task.LoopingCall(self.update) + self.tasks: list[AsyncioLoopingCall | LoopingCall] = [] + tsk = create_looping_call(self.update) self.tasks.append(tsk) tsk.start(self.check_interval, now=True) if self.limit: - tsk = task.LoopingCall(self._check_limit) + tsk = create_looping_call(self._check_limit) self.tasks.append(tsk) tsk.start(self.check_interval, now=True) if self.warning: - tsk = task.LoopingCall(self._check_warning) + tsk = create_looping_call(self._check_warning) self.tasks.append(tsk) tsk.start(self.check_interval, now=True) diff --git a/scrapy/extensions/periodic_log.py b/scrapy/extensions/periodic_log.py index f97577442..9158482fa 100644 --- a/scrapy/extensions/periodic_log.py +++ b/scrapy/extensions/periodic_log.py @@ -4,16 +4,20 @@ import logging from datetime import datetime, timezone from typing import TYPE_CHECKING, Any -from twisted.internet import task - from scrapy import Spider, signals from scrapy.exceptions import NotConfigured +from scrapy.utils.asyncio import ( + AsyncioLoopingCall, + create_looping_call, +) from scrapy.utils.serialize import ScrapyJSONEncoder if TYPE_CHECKING: - # typing.Self requires Python 3.11 from json import JSONEncoder + from twisted.internet.task import LoopingCall + + # typing.Self requires Python 3.11 from typing_extensions import Self from scrapy.crawler import Crawler @@ -37,7 +41,7 @@ class PeriodicLog: self.stats: StatsCollector = stats self.interval: float = interval self.multiplier: float = 60.0 / self.interval - self.task: task.LoopingCall | None = None + self.task: AsyncioLoopingCall | LoopingCall | None = None self.encoder: JSONEncoder = ScrapyJSONEncoder(sort_keys=True, indent=4) self.ext_stats_enabled: bool = bool(ext_stats) self.ext_stats_include: list[str] = ext_stats.get("include", []) @@ -97,7 +101,7 @@ class PeriodicLog: self.delta_prev: dict[str, int | float] = {} self.stats_prev: dict[str, int | float] = {} - self.task = task.LoopingCall(self.log) + self.task = create_looping_call(self.log) self.task.start(self.interval) def log(self) -> None: diff --git a/scrapy/utils/asyncio.py b/scrapy/utils/asyncio.py index a3f27bcc9..cae2dc033 100644 --- a/scrapy/utils/asyncio.py +++ b/scrapy/utils/asyncio.py @@ -3,22 +3,30 @@ from __future__ import annotations import asyncio +import logging +import time +from collections.abc import AsyncIterator, Callable, Coroutine, Iterable from typing import TYPE_CHECKING, Any, TypeVar +from twisted.internet.defer import Deferred +from twisted.internet.task import LoopingCall + from scrapy.utils.asyncgen import as_async_generator from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed if TYPE_CHECKING: - from collections.abc import AsyncIterator, Callable, Coroutine, Iterable - # typing.Concatenate and typing.ParamSpec require Python 3.10 from typing_extensions import Concatenate, ParamSpec _P = ParamSpec("_P") + _T = TypeVar("_T") +logger = logging.getLogger(__name__) + + def is_asyncio_available() -> bool: """Check if it's possible to call asyncio code that relies on the asyncio event loop. @@ -90,3 +98,97 @@ async def _parallel_asyncio( fill_task = asyncio.create_task(fill_queue()) work_tasks = [asyncio.create_task(worker()) for _ in range(count)] await asyncio.wait([fill_task, *work_tasks]) + + +class AsyncioLoopingCall: + """A simple implementation of a periodic call using asyncio, keeping + some API and behavior compatibility with the Twisted ``LoopingCall``. + + The function is called every *interval* seconds, independent of the finish + time of the previous call. If the function is still running when it's time + to call it again, calls are skipped until the function finishes. + + The function must not return a coroutine or a ``Deferred``. + """ + + def __init__(self, func: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs): + self._func: Callable[_P, _T] = func + self._args: tuple[Any, ...] = args + self._kwargs: dict[str, Any] = kwargs + self._task: asyncio.Task | None = None + self.interval: float | None = None + self._start_time: float | None = None + + @property + def running(self) -> bool: + return self._start_time is not None + + def start(self, interval: float, now: bool = True) -> None: + """Start calling the function every *interval* seconds. + + :param interval: The interval in seconds between calls. + :type interval: float + + :param now: If ``True``, also call the function immediately. + :type now: bool + """ + if self.running: + raise RuntimeError("AsyncioLoopingCall already running") + + if interval <= 0: + raise ValueError("Interval must be greater than 0") + + self.interval = interval + self._start_time = time.time() + if now: + self._call() + loop = asyncio.get_event_loop() + self._task = loop.create_task(self._loop()) + + def _to_sleep(self) -> float: + """Return the time to sleep until the next call.""" + assert self.interval is not None + assert self._start_time is not None + now = time.time() + running_for = now - self._start_time + return self.interval - (running_for % self.interval) + + async def _loop(self) -> None: + """Run an infinite loop that calls the function periodically.""" + while self.running: + await asyncio.sleep(self._to_sleep()) + self._call() + + def stop(self) -> None: + """Stop the periodic calls.""" + self.interval = self._start_time = None + if self._task is not None: + self._task.cancel() + self._task = None + + def _call(self) -> None: + """Execute the function.""" + try: + result = self._func(*self._args, **self._kwargs) + except Exception: + logger.exception("Error calling the AsyncioLoopingCall function") + self.stop() + else: + if isinstance(result, (Coroutine, Deferred)): + self.stop() + raise TypeError( + "The AsyncioLoopingCall function must not return a coroutine or a Deferred" + ) + + +def create_looping_call( + func: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs +) -> AsyncioLoopingCall | LoopingCall: + """Create an instance of a looping call class. + + This creates an instance of :class:`AsyncioLoopingCall` or + :class:`LoopingCall`, depending on whether asyncio support is available. + """ + if is_asyncio_available(): + return AsyncioLoopingCall(func, *args, **kwargs) + return LoopingCall(func, *args, **kwargs) diff --git a/tests/test_utils_asyncio.py b/tests/test_utils_asyncio.py index 6c47965a3..a6e52eb26 100644 --- a/tests/test_utils_asyncio.py +++ b/tests/test_utils_asyncio.py @@ -3,12 +3,18 @@ from __future__ import annotations import asyncio import random from typing import TYPE_CHECKING +from unittest import mock import pytest +from twisted.internet.defer import Deferred from twisted.trial import unittest from scrapy.utils.asyncgen import as_async_generator -from scrapy.utils.asyncio import _parallel_asyncio, is_asyncio_available +from scrapy.utils.asyncio import ( + AsyncioLoopingCall, + _parallel_asyncio, + is_asyncio_available, +) from scrapy.utils.defer import deferred_f_from_coro_f if TYPE_CHECKING: @@ -97,3 +103,43 @@ class TestParallelAsyncio(unittest.TestCase): ) assert list(range(length)) == sorted(results) assert max_parallel_count[0] <= self.CONCURRENT_ITEMS + + +@pytest.mark.only_asyncio +class TestAsyncioLoopingCall: + def test_looping_call(self): + func = mock.MagicMock() + looping_call = AsyncioLoopingCall(func) + looping_call.start(1, now=False) + assert looping_call.running + looping_call.stop() + assert not looping_call.running + assert not func.called + + def test_looping_call_now(self): + func = mock.MagicMock() + looping_call = AsyncioLoopingCall(func) + looping_call.start(1) + looping_call.stop() + assert func.called + + def test_looping_call_already_running(self): + looping_call = AsyncioLoopingCall(lambda: None) + looping_call.start(1) + with pytest.raises(RuntimeError): + looping_call.start(1) + looping_call.stop() + + def test_looping_call_interval(self): + looping_call = AsyncioLoopingCall(lambda: None) + with pytest.raises(ValueError, match="Interval must be greater than 0"): + looping_call.start(0) + with pytest.raises(ValueError, match="Interval must be greater than 0"): + looping_call.start(-1) + assert not looping_call.running + + def test_looping_call_bad_function(self): + looping_call = AsyncioLoopingCall(Deferred) + with pytest.raises(TypeError): + looping_call.start(0.1) + assert not looping_call.running