Add AsyncioLoopingCall. (#6855)

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", {}

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)

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(

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:

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)
 

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:

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)

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