diff --git a/scrapy/core/downloader/__init__.py b/scrapy/core/downloader/__init__.py index 4b5654826..c0833c84d 100644 --- a/scrapy/core/downloader/__init__.py +++ b/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 diff --git a/scrapy/core/downloader/middleware.py b/scrapy/core/downloader/middleware.py index a4055849d..fa45d62ee 100644 --- a/scrapy/core/downloader/middleware.py +++ b/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) diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index dc3a287b4..6b80ba9bf 100644 --- a/scrapy/core/scraper.py +++ b/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: diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py index 10aad7858..77fcaa357 100644 --- a/scrapy/core/spidermw.py +++ b/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: diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py index 888be81c3..55d9cae6c 100644 --- a/scrapy/pipelines/files.py +++ b/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 diff --git a/scrapy/pipelines/media.py b/scrapy/pipelines/media.py index e66b86ce6..04e1d14fa 100644 --- a/scrapy/pipelines/media.py +++ b/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]) - if eb: - d.addErrback(eb) - return d + yield _defer_sleep() + cached_result = info.downloaded[fp] + if isinstance(cached_result, Failure): + 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( - self.media_to_download, request, info, item=item - ) - dfd2: Deferred[FileInfo] = dfd.addCallback( - self._check_media_to_download, request, info, item=item - ) - dfd2.addErrback(self._log_exception) - dfd2.addBoth(self._cache_result_and_execute_waiters, fp, info) - return dfd2.addBoth(lambda _: wad) # it must return wad at last - - def _log_exception(self, result: Failure) -> Failure: - logger.exception(result) - return result + yield _defer_sleep() + result: FileInfo | Failure + try: + file_info = yield maybeDeferred( + self.media_to_download, request, info, item=item + ) + if file_info: + # got a result without downloading + result = file_info + else: + # download the result + result = yield self._check_media_to_download(request, info, item=item) + 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( - self, result: FileInfo | None, request: Request, info: SpiderInfo, item: Any - ) -> FileInfo | Deferred[FileInfo]: - if result is not None: - return result - dfd: Deferred[Response] - if self.download_func: - # this ugly code was left only to support tests. TODO: remove - dfd = mustbe_deferred(self.download_func, request, info.spider) - else: - self._modify_media_request(request) - assert self.crawler.engine - dfd = self.crawler.engine.download(request) - dfd2: Deferred[FileInfo] = dfd.addCallback( - self.media_downloaded, request, info, item=item - ) - dfd2.addErrback(self.media_failed, request, info) - return dfd2 + @inlineCallbacks + def _check_media_to_download( # pylint: disable=inconsistent-return-statements + self, request: Request, info: SpiderInfo, item: Any + ) -> Generator[Deferred[Any], Any, FileInfo]: + try: + if self.download_func: + # this ugly code was left only to support tests. TODO: remove + response = yield maybeDeferred(self.download_func, request, info.spider) + else: + self._modify_media_request(request) + assert self.crawler.engine + response = yield self.crawler.engine.download(request) + return self.media_downloaded(response, request, info, item=item) + except Exception: + failure = self.media_failed(Failure(), request, info) + if isinstance(failure, Failure): + warnings.warn( + 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 diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index 2df82c8f2..50f2af5b7 100644 --- a/scrapy/utils/defer.py +++ b/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.utils.asyncio import is_asyncio_available +from scrapy.exceptions import ScrapyDeprecationWarning +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.""" - from twisted.internet import reactor - + """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. + """ 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) - return defer_succeed(result) + reactor.callLater(_DEFER_DELAY, d.errback, 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) diff --git a/tests/test_pipeline_media.py b/tests/test_pipeline_media.py index 40149f184..5364b09a8 100644 --- a/tests/test_pipeline_media.py +++ b/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", + ] diff --git a/tests/test_spidermiddleware.py b/tests/test_spidermiddleware.py index 00b90bc35..8d9f3e83d 100644 --- a/tests/test_spidermiddleware.py +++ b/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): - for item in super()._scrape_func(): + async def _callback(self) -> Any: + 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): - for item in super()._scrape_func(): + async def _callback(self) -> Any: + 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: diff --git a/tests/test_utils_defer.py b/tests/test_utils_defer.py index 36533ad5d..b31d8c3d9 100644 --- a/tests/test_utils_defer.py +++ b/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]: