Deprecate mustbe_deferred() and defer_*(). (#6937)

scrapy/core/downloader/__init__.py

@@ -21,9 +21,9 @@ from scrapy.utils.asyncio import (
     create_looping_call,
 )
 from scrapy.utils.defer import (
+    _defer_sleep_async,
     deferred_from_coro,
     maybe_deferred_to_future,
-    mustbe_deferred,
 )
 from scrapy.utils.httpobj import urlparse_cached
 
@@ -231,7 +231,7 @@ class Downloader:
         try:
             # 1. Download the response
             response: Response = await maybe_deferred_to_future(
-                mustbe_deferred(self.handlers.download_request, request, spider)
+                self.handlers.download_request(request, spider)
             )
             # 2. Notify response_downloaded listeners about the recent download
             # before querying queue for next request
@@ -242,6 +242,9 @@ class Downloader:
                 spider=spider,
             )
             return response
+        except Exception:
+            await _defer_sleep_async()
+            raise
         finally:
             # 3. After response arrives, remove the request from transferring
             # state to free up the transferring slot so it can be used by the

scrapy/core/downloader/middleware.py

@@ -15,7 +15,7 @@ from scrapy.exceptions import _InvalidOutput
 from scrapy.http import Request, Response
 from scrapy.middleware import MiddlewareManager
 from scrapy.utils.conf import build_component_list
-from scrapy.utils.defer import deferred_from_coro, mustbe_deferred
+from scrapy.utils.defer import _defer_sleep, deferred_from_coro
 
 if TYPE_CHECKING:
     from collections.abc import Generator
@@ -110,8 +110,9 @@ class DownloaderMiddlewareManager(MiddlewareManager):
             raise exception
 
         try:
-            result: Response | Request = yield mustbe_deferred(process_request, request)
+            result: Response | Request = yield process_request(request)
         except Exception as ex:
+            yield _defer_sleep()
             # either returns a request or response (which we pass to process_response())
             # or reraises the exception
             result = yield process_exception(ex)

scrapy/core/scraper.py

@@ -23,7 +23,7 @@ from scrapy.exceptions import (
 from scrapy.http import Request, Response
 from scrapy.utils.asyncio import _parallel_asyncio, is_asyncio_available
 from scrapy.utils.defer import (
-    _defer_sleep,
+    _defer_sleep_async,
     aiter_errback,
     deferred_f_from_coro_f,
     deferred_from_coro,
@@ -236,7 +236,7 @@ class Scraper:
         self, result: Response | Failure, request: Request
     ) -> Iterable[Any] | AsyncIterator[Any]:
         """Call the request callback or errback with the response or failure."""
-        await maybe_deferred_to_future(_defer_sleep())
+        await _defer_sleep_async()
         assert self.crawler.spider
         if isinstance(result, Response):
             if getattr(result, "request", None) is None:

scrapy/core/spidermw.py

@@ -23,9 +23,9 @@ from scrapy.middleware import MiddlewareManager
 from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.conf import build_component_list
 from scrapy.utils.defer import (
+    _defer_sleep_async,
     deferred_from_coro,
     maybe_deferred_to_future,
-    mustbe_deferred,
 )
 from scrapy.utils.python import MutableAsyncChain, MutableChain, global_object_name
 
@@ -390,12 +390,11 @@ class SpiderMiddlewareManager(MiddlewareManager):
 
         try:
             it: Iterable[_T] | AsyncIterator[_T] = await maybe_deferred_to_future(
-                mustbe_deferred(
+                self._process_spider_input(scrape_func, response, request, spider)
-                    self._process_spider_input, scrape_func, response, request, spider
-                )
             )
             return await process_callback_output(it)
         except Exception as ex:
+            await _defer_sleep_async()
             return process_spider_exception(ex)
 
     async def process_start(self, spider: Spider) -> AsyncIterator[Any] | None:

scrapy/pipelines/files.py

@@ -562,7 +562,7 @@ class FilesPipeline(MediaPipeline):
 
     def media_to_download(
         self, request: Request, info: MediaPipeline.SpiderInfo, *, item: Any = None
-    ) -> Deferred[FileInfo | None]:
+    ) -> Deferred[FileInfo | None] | None:
         def _onsuccess(result: StatInfo) -> FileInfo | None:
             if not result:
                 return None  # returning None force download

scrapy/pipelines/media.py

@@ -5,24 +5,30 @@ import logging
 import warnings
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from typing import TYPE_CHECKING, Any, Literal, NoReturn, TypedDict, Union, cast
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, Union, cast
 
 from twisted import version as twisted_version
-from twisted.internet.defer import Deferred, DeferredList
+from twisted.internet.defer import (
+    Deferred,
+    DeferredList,
+    inlineCallbacks,
+    maybeDeferred,
+)
 from twisted.python.failure import Failure
 from twisted.python.versions import Version
 
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.http.request import NO_CALLBACK, Request
 from scrapy.settings import Settings
+from scrapy.utils.asyncio import call_later
 from scrapy.utils.datatypes import SequenceExclude
-from scrapy.utils.defer import defer_result, mustbe_deferred
+from scrapy.utils.defer import _DEFER_DELAY, _defer_sleep
 from scrapy.utils.log import failure_to_exc_info
 from scrapy.utils.misc import arg_to_iter
 from scrapy.utils.python import get_func_args, global_object_name
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Generator
 
     # typing.Self requires Python 3.11
     from typing_extensions import Self
@@ -173,20 +179,25 @@ class MediaPipeline(ABC):
         )
         return dfd.addCallback(self.item_completed, item, info)
 
+    @inlineCallbacks
     def _process_request(
         self, request: Request, info: SpiderInfo, item: Any
-    ) -> Deferred[FileInfo]:
+    ) -> Generator[Deferred[Any], Any, FileInfo]:
         fp = self._fingerprinter.fingerprint(request)
+
         eb = request.errback
         request.callback = NO_CALLBACK
         request.errback = None
 
         # Return cached result if request was already seen
         if fp in info.downloaded:
-            d = defer_result(info.downloaded[fp])
+            yield _defer_sleep()
-            if eb:
+            cached_result = info.downloaded[fp]
-                d.addErrback(eb)
+            if isinstance(cached_result, Failure):
-            return d
+                if eb:
+                    return eb(cached_result)
+                cached_result.raiseException()
+            return cached_result
 
         # Otherwise, wait for result
         wad: Deferred[FileInfo] = Deferred()
@@ -196,23 +207,27 @@ class MediaPipeline(ABC):
 
         # Check if request is downloading right now to avoid doing it twice
         if fp in info.downloading:
-            return wad
+            return (yield wad)
 
         # Download request checking media_to_download hook output first
         info.downloading.add(fp)
-        dfd: Deferred[FileInfo | None] = mustbe_deferred(
+        yield _defer_sleep()
-            self.media_to_download, request, info, item=item
+        result: FileInfo | Failure
-        )
+        try:
-        dfd2: Deferred[FileInfo] = dfd.addCallback(
+            file_info = yield maybeDeferred(
-            self._check_media_to_download, request, info, item=item
+                self.media_to_download, request, info, item=item
-        )
+            )
-        dfd2.addErrback(self._log_exception)
+            if file_info:
-        dfd2.addBoth(self._cache_result_and_execute_waiters, fp, info)
+                # got a result without downloading
-        return dfd2.addBoth(lambda _: wad)  # it must return wad at last
+                result = file_info
-
+            else:
-    def _log_exception(self, result: Failure) -> Failure:
+                # download the result
-        logger.exception(result)
+                result = yield self._check_media_to_download(request, info, item=item)
-        return result
+        except Exception:
+            result = Failure()
+            logger.exception(result)
+        self._cache_result_and_execute_waiters(result, fp, info)
+        return (yield wad)  # it must return wad at last
 
     def _modify_media_request(self, request: Request) -> None:
         if self.handle_httpstatus_list:
@@ -220,24 +235,29 @@ class MediaPipeline(ABC):
         else:
             request.meta["handle_httpstatus_all"] = True
 
-    def _check_media_to_download(
+    @inlineCallbacks
-        self, result: FileInfo | None, request: Request, info: SpiderInfo, item: Any
+    def _check_media_to_download(  # pylint: disable=inconsistent-return-statements
-    ) -> FileInfo | Deferred[FileInfo]:
+        self, request: Request, info: SpiderInfo, item: Any
-        if result is not None:
+    ) -> Generator[Deferred[Any], Any, FileInfo]:
-            return result
+        try:
-        dfd: Deferred[Response]
+            if self.download_func:
-        if self.download_func:
+                # this ugly code was left only to support tests. TODO: remove
-            # this ugly code was left only to support tests. TODO: remove
+                response = yield maybeDeferred(self.download_func, request, info.spider)
-            dfd = mustbe_deferred(self.download_func, request, info.spider)
+            else:
-        else:
+                self._modify_media_request(request)
-            self._modify_media_request(request)
+                assert self.crawler.engine
-            assert self.crawler.engine
+                response = yield self.crawler.engine.download(request)
-            dfd = self.crawler.engine.download(request)
+            return self.media_downloaded(response, request, info, item=item)
-        dfd2: Deferred[FileInfo] = dfd.addCallback(
+        except Exception:
-            self.media_downloaded, request, info, item=item
+            failure = self.media_failed(Failure(), request, info)
-        )
+            if isinstance(failure, Failure):
-        dfd2.addErrback(self.media_failed, request, info)
+                warnings.warn(
-        return dfd2
+                    f"{global_object_name(self.media_failed)} returned a Failure instance."
+                    f" This is deprecated, please raise an exception instead, e.g. via failure.raiseException().",
+                    category=ScrapyDeprecationWarning,
+                    stacklevel=2,
+                )
+                failure.raiseException()
 
     def _cache_result_and_execute_waiters(
         self, result: FileInfo | Failure, fp: bytes, info: SpiderInfo
@@ -273,13 +293,16 @@ class MediaPipeline(ABC):
         info.downloading.remove(fp)
         info.downloaded[fp] = result  # cache result
         for wad in info.waiting.pop(fp):
-            defer_result(result).chainDeferred(wad)
+            if isinstance(result, Failure):
+                call_later(_DEFER_DELAY, wad.errback, result)
+            else:
+                call_later(_DEFER_DELAY, wad.callback, result)
 
     # Overridable Interface
     @abstractmethod
     def media_to_download(
         self, request: Request, info: SpiderInfo, *, item: Any = None
-    ) -> Deferred[FileInfo | None]:
+    ) -> Deferred[FileInfo | None] | None:
         """Check request before starting download"""
         raise NotImplementedError
 
@@ -303,7 +326,7 @@ class MediaPipeline(ABC):
     @abstractmethod
     def media_failed(
         self, failure: Failure, request: Request, info: SpiderInfo
-    ) -> NoReturn:
+    ) -> Failure:
         """Handler for failed downloads"""
         raise NotImplementedError
 

scrapy/utils/defer.py

@@ -21,8 +21,8 @@ from twisted.internet.defer import (
 from twisted.internet.task import Cooperator
 from twisted.python import failure
 
-from scrapy.exceptions import IgnoreRequest, ScrapyDeprecationWarning
+from scrapy.exceptions import ScrapyDeprecationWarning
-from scrapy.utils.asyncio import is_asyncio_available
+from scrapy.utils.asyncio import call_later, is_asyncio_available
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Callable
@@ -49,6 +49,13 @@ def defer_fail(_failure: Failure) -> Deferred[Any]:
     It delays by 100ms so reactor has a chance to go through readers and writers
     before attending pending delayed calls, so do not set delay to zero.
     """
+    warnings.warn(
+        "scrapy.utils.defer.defer_fail() is deprecated, use"
+        " twisted.internet.defer.fail(), plus an explicit sleep if needed.",
+        category=ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
+
     from twisted.internet import reactor
 
     d: Deferred[Any] = Deferred()
@@ -63,6 +70,13 @@ def defer_succeed(result: _T) -> Deferred[_T]:
     It delays by 100ms so reactor has a chance to go through readers and writers
     before attending pending delayed calls, so do not set delay to zero.
     """
+    warnings.warn(
+        "scrapy.utils.defer.defer_succeed() is deprecated, use"
+        " twisted.internet.defer.succeed(), plus an explicit sleep if needed.",
+        category=ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
+
     from twisted.internet import reactor
 
     d: Deferred[_T] = Deferred()
@@ -71,20 +85,44 @@ def defer_succeed(result: _T) -> Deferred[_T]:
 
 
 def _defer_sleep() -> Deferred[None]:
-    """Like ``defer_succeed`` and ``defer_fail`` but doesn't call any real callbacks."""
+    """Delay by _DEFER_DELAY so reactor has a chance to go through readers and writers
-    from twisted.internet import reactor
+    before attending pending delayed calls, so do not set delay to zero.
-
+    """
     d: Deferred[None] = Deferred()
-    reactor.callLater(_DEFER_DELAY, d.callback, None)
+    call_later(_DEFER_DELAY, d.callback, None)
     return d
 
 
+async def _defer_sleep_async() -> None:
+    """Delay by _DEFER_DELAY so reactor has a chance to go through readers and writers
+    before attending pending delayed calls, so do not set delay to zero.
+    """
+    if is_asyncio_available():
+        await asyncio.sleep(_DEFER_DELAY)
+    else:
+        await _defer_sleep()
+
+
 def defer_result(result: Any) -> Deferred[Any]:
+    warnings.warn(
+        "scrapy.utils.defer.defer_result() is deprecated, use"
+        " twisted.internet.defer.success() and twisted.internet.defer.fail(),"
+        " plus an explicit sleep if needed, or explicit reactor.callLater().",
+        category=ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
+
     if isinstance(result, Deferred):
         return result
+
+    from twisted.internet import reactor
+
+    d: Deferred[Any] = Deferred()
     if isinstance(result, failure.Failure):
-        return defer_fail(result)
+        reactor.callLater(_DEFER_DELAY, d.errback, result)
-    return defer_succeed(result)
+    else:
+        reactor.callLater(_DEFER_DELAY, d.callback, result)
+    return d
 
 
 @overload
@@ -93,14 +131,6 @@ def mustbe_deferred(
 ) -> Deferred[_T]: ...
 
 
-@overload
-def mustbe_deferred(
-    f: Callable[_P, Coroutine[Deferred[Any], Any, _T]],
-    *args: _P.args,
-    **kw: _P.kwargs,
-) -> Deferred[_T]: ...
-
-
 @overload
 def mustbe_deferred(
     f: Callable[_P, _T], *args: _P.args, **kw: _P.kwargs
@@ -108,22 +138,24 @@ def mustbe_deferred(
 
 
 def mustbe_deferred(
-    f: Callable[_P, Deferred[_T] | Coroutine[Deferred[Any], Any, _T] | _T],
+    f: Callable[_P, Deferred[_T] | _T],
     *args: _P.args,
     **kw: _P.kwargs,
 ) -> Deferred[_T]:
     """Same as twisted.internet.defer.maybeDeferred, but delay calling
     callback/errback to next reactor loop
     """
+    warnings.warn(
+        "scrapy.utils.defer.mustbe_deferred() is deprecated, use"
+        " twisted.internet.defer.maybeDeferred(), with an explicit sleep if needed.",
+        category=ScrapyDeprecationWarning,
+        stacklevel=2,
+    )
+    result: _T | Deferred[_T] | Failure
     try:
         result = f(*args, **kw)
-    # FIXME: Hack to avoid introspecting tracebacks. This to speed up
-    # processing of IgnoreRequest errors which are, by far, the most common
-    # exception in Scrapy - see #125
-    except IgnoreRequest as e:
-        return defer_fail(failure.Failure(e))
     except Exception:
-        return defer_fail(failure.Failure())
+        result = failure.Failure()
     return defer_result(result)
 
 

tests/test_pipeline_media.py

@@ -9,6 +9,7 @@ from twisted.python.failure import Failure
 from twisted.trial import unittest
 
 from scrapy import signals
+from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.http import Request, Response
 from scrapy.http.request import NO_CALLBACK
 from scrapy.pipelines.files import FileException
@@ -36,7 +37,7 @@ class UserDefinedPipeline(MediaPipeline):
         return {}
 
     def media_failed(self, failure, request, info):
-        return failure
+        failure.raiseException()
 
     def file_path(self, request, response=None, info=None, *, item=None):
         return ""
@@ -227,7 +228,8 @@ class TestMediaPipeline(TestBaseMediaPipeline):
     @inlineCallbacks
     def test_result_failure(self):
         self.pipe.LOG_FAILED_RESULTS = False
-        fail = Failure(Exception())
+        exc = Exception("foo")
+        fail = Failure(exc)
         req = Request(
             "http://url1",
             meta={"response": fail},
@@ -235,7 +237,10 @@ class TestMediaPipeline(TestBaseMediaPipeline):
         )
         item = {"requests": req}
         new_item = yield self.pipe.process_item(item, self.spider)
-        assert new_item["results"] == [(False, fail)]
+        assert len(new_item["results"]) == 1
+        assert new_item["results"][0][0] is False
+        assert isinstance(new_item["results"][0][1], Failure)
+        assert new_item["results"][0][1].value == exc
         assert self.pipe._mockcalled == [
             "get_media_requests",
             "media_to_download",
@@ -249,11 +254,16 @@ class TestMediaPipeline(TestBaseMediaPipeline):
         self.pipe.LOG_FAILED_RESULTS = False
         rsp1 = Response("http://url1")
         req1 = Request("http://url1", meta={"response": rsp1})
-        fail = Failure(Exception())
+        exc = Exception("foo")
+        fail = Failure(exc)
         req2 = Request("http://url2", meta={"response": fail})
         item = {"requests": [req1, req2]}
         new_item = yield self.pipe.process_item(item, self.spider)
-        assert new_item["results"] == [(True, {}), (False, fail)]
+        assert len(new_item["results"]) == 2
+        assert new_item["results"][0] == (True, {})
+        assert new_item["results"][1][0] is False
+        assert isinstance(new_item["results"][1][1], Failure)
+        assert new_item["results"][1][1].value == exc
         m = self.pipe._mockcalled
         # only once
         assert m[0] == "get_media_requests"  # first hook called
@@ -532,3 +542,62 @@ class TestBuildFromCrawler:
             assert pipe._fingerprinter
             assert len(w) == 0
             assert pipe._from_crawler_called
+
+
+class MediaFailedFailurePipeline(MockedMediaPipeline):
+    def media_failed(self, failure, request, info):
+        self._mockcalled.append("media_failed")
+        return failure  # deprecated
+
+
+class TestMediaFailedFailure(unittest.TestCase):
+    """Test that media_failed() can return a failure instead of raising."""
+
+    pipeline_class = MediaFailedFailurePipeline
+    settings = None
+
+    def setUp(self):
+        spider_cls = Spider
+        self.spider = spider_cls("media.com")
+        crawler = get_crawler(spider_cls, self.settings)
+        self.pipe = self.pipeline_class.from_crawler(crawler)
+        self.pipe.download_func = _mocked_download_func
+        self.pipe.open_spider(self.spider)
+        self.info = self.pipe.spiderinfo
+        self.fingerprint = crawler.request_fingerprinter.fingerprint
+
+    def tearDown(self):
+        for name, signal in vars(signals).items():
+            if not name.startswith("_"):
+                disconnect_all(signal)
+
+    def _errback(self, result):
+        self.pipe._mockcalled.append("request_errback")
+        return result
+
+    @inlineCallbacks
+    def test_result_failure(self):
+        self.pipe.LOG_FAILED_RESULTS = False
+        exc = Exception("foo")
+        fail = Failure(exc)
+        req = Request(
+            "http://url1",
+            meta={"response": fail},
+            errback=self._errback,
+        )
+        item = {"requests": req}
+        with pytest.warns(
+            ScrapyDeprecationWarning, match="media_failed returned a Failure instance"
+        ):
+            new_item = yield self.pipe.process_item(item, self.spider)
+        assert len(new_item["results"]) == 1
+        assert new_item["results"][0][0] is False
+        assert isinstance(new_item["results"][0][1], Failure)
+        assert new_item["results"][0][1].value == exc
+        assert self.pipe._mockcalled == [
+            "get_media_requests",
+            "media_to_download",
+            "media_failed",
+            "request_errback",
+            "item_completed",
+        ]

tests/test_spidermiddleware.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import AsyncIterator, Iterable
 from inspect import isasyncgen
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from unittest import mock
 
 import pytest
@@ -18,6 +18,9 @@ from scrapy.utils.asyncgen import collect_asyncgen
 from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future
 from scrapy.utils.test import get_crawler
 
+if TYPE_CHECKING:
+    from twisted.python.failure import Failure
+
 
 class TestSpiderMiddleware(TestCase):
     def setUp(self):
@@ -31,7 +34,13 @@ class TestSpiderMiddleware(TestCase):
         """Execute spider mw manager's scrape_response method and return the result.
         Raise exception in case of failure.
         """
-        scrape_func = mock.MagicMock()
+
+        def scrape_func(
+            response: Response | Failure, request: Request
+        ) -> defer.Deferred[Iterable[Any]]:
+            it = mock.MagicMock()
+            return defer.succeed(it)
+
         return await maybe_deferred_to_future(
             self.mwman.scrape_response(
                 scrape_func, self.response, self.request, self.spider
@@ -122,11 +131,16 @@ class TestBaseAsyncSpiderMiddleware(TestSpiderMiddleware):
             start_index = 10
         return {i: c for c, i in enumerate(mw_classes, start=start_index)}
 
-    def _scrape_func(self, *args, **kwargs):
+    def _callback(self) -> Any:
         yield {"foo": 1}
         yield {"foo": 2}
         yield {"foo": 3}
 
+    def _scrape_func(
+        self, response: Response | Failure, request: Request
+    ) -> defer.Deferred[Iterable[Any] | AsyncIterator[Any]]:
+        return defer.succeed(self._callback())
+
     async def _get_middleware_result(
         self, *mw_classes: type[Any], start_index: int | None = None
     ) -> Any:
@@ -272,8 +286,8 @@ class TestProcessSpiderOutputSimple(TestBaseAsyncSpiderMiddleware):
 class TestProcessSpiderOutputAsyncGen(TestProcessSpiderOutputSimple):
     """process_spider_output tests for async generator callbacks"""
 
-    async def _scrape_func(self, *args, **kwargs):
+    async def _callback(self) -> Any:
-        for item in super()._scrape_func():
+        for item in super()._callback():
             yield item
 
     @deferred_f_from_coro_f
@@ -503,8 +517,8 @@ class TestBuiltinMiddlewareSimple(TestBaseAsyncSpiderMiddleware):
 
 
 class TestBuiltinMiddlewareAsyncGen(TestBuiltinMiddlewareSimple):
-    async def _scrape_func(self, *args, **kwargs):
+    async def _callback(self) -> Any:
-        for item in super()._scrape_func():
+        for item in super()._callback():
             yield item
 
     @deferred_f_from_coro_f
@@ -546,7 +560,7 @@ class TestProcessSpiderException(TestBaseAsyncSpiderMiddleware):
     MW_EXC_SIMPLE = ProcessSpiderExceptionSimpleIterableMiddleware
     MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIteratorMiddleware
 
-    def _scrape_func(self, *args, **kwargs):
+    def _callback(self) -> Any:
         1 / 0
 
     async def _test_asyncgen_nodowngrade(self, *mw_classes: type[Any]) -> None:

tests/test_utils_defer.py

@@ -28,6 +28,7 @@ if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
 
 
+@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
 class TestMustbeDeferred(unittest.TestCase):
     @inlineCallbacks
     def test_success_function(self) -> Generator[Deferred[Any], Any, None]: