diff --git a/docs/news.rst b/docs/news.rst index 96e5cb2b1..19d990a50 100644 --- a/docs/news.rst +++ b/docs/news.rst @@ -74,6 +74,11 @@ Backward-incompatible changes :class:`~scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware` or calls its methods directly. +- :meth:`scrapy.pipelines.media.MediaPipeline.process_item` now returns a + coroutine, previously it returned a + :class:`~twisted.internet.defer.Deferred` object. This + change only impacts code that calls this method directly. + .. _release-2.13.4: Scrapy 2.13.4 (2025-11-17) diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py index 25af0aa72..ad4abbbb6 100644 --- a/scrapy/pipelines/files.py +++ b/scrapy/pipelines/files.py @@ -12,6 +12,7 @@ import hashlib import logging import mimetypes import time +import warnings from collections import defaultdict from contextlib import suppress from ftplib import FTP @@ -24,7 +25,7 @@ from itemadapter import ItemAdapter from twisted.internet.defer import Deferred, maybeDeferred from twisted.internet.threads import deferToThread -from scrapy.exceptions import IgnoreRequest, NotConfigured +from scrapy.exceptions import IgnoreRequest, NotConfigured, ScrapyDeprecationWarning from scrapy.http import Request, Response from scrapy.http.request import NO_CALLBACK from scrapy.pipelines.media import FileInfo, FileInfoOrError, MediaPipeline @@ -36,7 +37,6 @@ from scrapy.utils.python import to_bytes from scrapy.utils.request import referer_str if TYPE_CHECKING: - from collections.abc import Callable from os import PathLike from twisted.python.failure import Failure @@ -44,7 +44,6 @@ if TYPE_CHECKING: # typing.Self requires Python 3.11 from typing_extensions import Self - from scrapy import Spider from scrapy.crawler import Crawler from scrapy.settings import BaseSettings @@ -443,10 +442,18 @@ class FilesPipeline(MediaPipeline): def __init__( self, store_uri: str | PathLike[str], - download_func: Callable[[Request, Spider], Response] | None = None, + download_func: None = None, *, crawler: Crawler, ): + if download_func is not None: # pragma: no cover + warnings.warn( + "The download_func argument of FilesPipeline.__init__() is ignored" + " and will be removed in a future Scrapy version.", + category=ScrapyDeprecationWarning, + stacklevel=2, + ) + if not (store_uri and (store_uri := _to_string(store_uri))): from scrapy.pipelines.images import ImagesPipeline # noqa: PLC0415 @@ -476,7 +483,7 @@ class FilesPipeline(MediaPipeline): resolve("FILES_RESULT_FIELD"), self.FILES_RESULT_FIELD ) - super().__init__(download_func=download_func, crawler=crawler) + super().__init__(crawler=crawler) @classmethod def from_crawler(cls, crawler: Crawler) -> Self: diff --git a/scrapy/pipelines/images.py b/scrapy/pipelines/images.py index 687d29fa5..f5d1ad1b9 100644 --- a/scrapy/pipelines/images.py +++ b/scrapy/pipelines/images.py @@ -8,20 +8,21 @@ from __future__ import annotations import functools import hashlib +import warnings from contextlib import suppress from io import BytesIO from typing import TYPE_CHECKING, Any from itemadapter import ItemAdapter -from scrapy.exceptions import NotConfigured +from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning from scrapy.http import Request, Response from scrapy.http.request import NO_CALLBACK from scrapy.pipelines.files import FileException, FilesPipeline, _md5sum from scrapy.utils.python import to_bytes if TYPE_CHECKING: - from collections.abc import Callable, Iterable + from collections.abc import Iterable from os import PathLike from PIL import Image @@ -29,7 +30,6 @@ if TYPE_CHECKING: # typing.Self requires Python 3.11 from typing_extensions import Self - from scrapy import Spider from scrapy.crawler import Crawler from scrapy.pipelines.media import FileInfoOrError, MediaPipeline @@ -55,10 +55,18 @@ class ImagesPipeline(FilesPipeline): def __init__( self, store_uri: str | PathLike[str], - download_func: Callable[[Request, Spider], Response] | None = None, + download_func: None = None, *, crawler: Crawler, ): + if download_func is not None: # pragma: no cover + warnings.warn( + "The download_func argument of ImagesPipeline.__init__() is ignored" + " and will be removed in a future Scrapy version.", + category=ScrapyDeprecationWarning, + stacklevel=2, + ) + try: from PIL import Image, ImageOps # noqa: PLC0415 @@ -69,7 +77,7 @@ class ImagesPipeline(FilesPipeline): "ImagesPipeline requires installing Pillow 8.3.2 or later" ) - super().__init__(store_uri, download_func=download_func, crawler=crawler) + super().__init__(store_uri, crawler=crawler) settings = crawler.settings resolve = functools.partial( diff --git a/scrapy/pipelines/media.py b/scrapy/pipelines/media.py index 438e5e7d6..cea6b9c35 100644 --- a/scrapy/pipelines/media.py +++ b/scrapy/pipelines/media.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import functools import logging import warnings @@ -8,28 +9,27 @@ from collections import defaultdict from typing import TYPE_CHECKING, Any, Literal, TypeAlias, TypedDict, cast from twisted import version as twisted_version -from twisted.internet.defer import ( - Deferred, - DeferredList, - inlineCallbacks, - maybeDeferred, -) +from twisted.internet.defer import Deferred, DeferredList 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.utils.asyncio import call_later +from scrapy.utils.asyncio import call_later, is_asyncio_available from scrapy.utils.datatypes import SequenceExclude from scrapy.utils.decorators import _warn_spider_arg -from scrapy.utils.defer import _DEFER_DELAY, _defer_sleep, deferred_from_coro +from scrapy.utils.defer import ( + _DEFER_DELAY, + _defer_sleep_async, + deferred_from_coro, + ensure_awaitable, + maybe_deferred_to_future, +) from scrapy.utils.log import failure_to_exc_info from scrapy.utils.misc import arg_to_iter from scrapy.utils.python import global_object_name if TYPE_CHECKING: - from collections.abc import Callable, Generator - # typing.Self requires Python 3.11 from typing_extensions import Self @@ -68,16 +68,22 @@ class MediaPipeline(ABC): def __init__( self, - download_func: Callable[[Request, Spider], Response] | None = None, + download_func: None = None, *, crawler: Crawler, ): + if download_func is not None: # pragma: no cover + warnings.warn( + "The download_func argument of MediaPipeline.__init__() is ignored" + " and will be removed in a future Scrapy version.", + category=ScrapyDeprecationWarning, + stacklevel=2, + ) self.crawler: Crawler = crawler assert crawler.request_fingerprinter self._fingerprinter: RequestFingerprinterProtocol = ( crawler.request_fingerprinter ) - self.download_func = download_func settings = crawler.settings resolve = functools.partial( @@ -119,21 +125,31 @@ class MediaPipeline(ABC): self.spiderinfo = self.SpiderInfo(self.crawler.spider) @_warn_spider_arg - def process_item( - self, item: Any, spider: Spider | None = None - ) -> Deferred[list[FileInfoOrError]]: + async def process_item(self, item: Any, spider: Spider | None = None) -> Any: info = self.spiderinfo requests = arg_to_iter(self.get_media_requests(item, info)) - dlist = [self._process_request(r, info, item) for r in requests] - dfd = cast( - "Deferred[list[FileInfoOrError]]", DeferredList(dlist, consumeErrors=True) - ) - return dfd.addCallback(self.item_completed, item, info) + coros = [self._process_request(r, info, item) for r in requests] + results: list[FileInfoOrError] = [] + if coros: + if is_asyncio_available(): + results_asyncio = await asyncio.gather(*coros, return_exceptions=True) + for res in results_asyncio: + if isinstance(res, BaseException): + results.append((False, Failure(res))) + else: + results.append((True, res)) + else: + results = await cast( + "Deferred[list[FileInfoOrError]]", + DeferredList( + (deferred_from_coro(coro) for coro in coros), consumeErrors=True + ), + ) + return self.item_completed(results, item, info) - @inlineCallbacks - def _process_request( + async def _process_request( self, request: Request, info: SpiderInfo, item: Any - ) -> Generator[Deferred[Any], Any, FileInfo]: + ) -> FileInfo: fp = self._fingerprinter.fingerprint(request) eb = request.errback @@ -142,7 +158,7 @@ class MediaPipeline(ABC): # Return cached result if request was already seen if fp in info.downloaded: - yield _defer_sleep() + await _defer_sleep_async() cached_result = info.downloaded[fp] if isinstance(cached_result, Failure): if eb: @@ -158,27 +174,27 @@ class MediaPipeline(ABC): # Check if request is downloading right now to avoid doing it twice if fp in info.downloading: - return (yield wad) + return await maybe_deferred_to_future(wad) # Download request checking media_to_download hook output first info.downloading.add(fp) - yield _defer_sleep() + await _defer_sleep_async() result: FileInfo | Failure try: - file_info = yield maybeDeferred( - self.media_to_download, request, info, item=item + file_info: FileInfo | None = await ensure_awaitable( + 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) + result = await 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 + return await maybe_deferred_to_future(wad) # it must return wad at last def _modify_media_request(self, request: Request) -> None: if self.handle_httpstatus_list: @@ -186,20 +202,13 @@ class MediaPipeline(ABC): else: request.meta["handle_httpstatus_all"] = True - @inlineCallbacks - def _check_media_to_download( # pylint: disable=inconsistent-return-statements + async def _check_media_to_download( self, request: Request, info: SpiderInfo, item: Any - ) -> Generator[Deferred[Any], Any, FileInfo]: + ) -> 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 deferred_from_coro( - self.crawler.engine.download_async(request) - ) + self._modify_media_request(request) + assert self.crawler.engine + response = await self.crawler.engine.download_async(request) return self.media_downloaded(response, request, info, item=item) except Exception: failure = self.media_failed(Failure(), request, info) diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index c27f8bb34..2a09f99bb 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -26,7 +26,7 @@ from twisted.internet.task import Cooperator from twisted.python import failure from scrapy.exceptions import ScrapyDeprecationWarning -from scrapy.utils.asyncio import call_later, is_asyncio_available +from scrapy.utils.asyncio import is_asyncio_available if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable @@ -84,15 +84,6 @@ def defer_succeed(result: _T) -> Deferred[_T]: return d -def _defer_sleep() -> Deferred[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. - """ - d: Deferred[None] = Deferred() - 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. @@ -100,7 +91,11 @@ async def _defer_sleep_async() -> None: if is_asyncio_available(): await asyncio.sleep(_DEFER_DELAY) else: - await _defer_sleep() + from twisted.internet import reactor + + d: Deferred[None] = Deferred() + reactor.callLater(_DEFER_DELAY, d.callback, None) + await d def defer_result(result: Any) -> Deferred[Any]: diff --git a/tests/test_pipeline_files.py b/tests/test_pipeline_files.py index 966cb320e..6ab5d9a8e 100644 --- a/tests/test_pipeline_files.py +++ b/tests/test_pipeline_files.py @@ -13,6 +13,7 @@ from shutil import rmtree from tempfile import mkdtemp from typing import Any from unittest import mock +from unittest.mock import MagicMock from urllib.parse import urlparse import attr @@ -31,6 +32,7 @@ from scrapy.pipelines.files import ( S3FilesStore, ) from scrapy.settings import Settings +from scrapy.utils.defer import deferred_f_from_coro_f from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler from tests.mockserver.ftp import MockFTPServer @@ -83,8 +85,8 @@ class TestFilesPipeline: settings_dict = {"FILES_STORE": self.tempdir} crawler = get_crawler(DefaultSpider, settings_dict=settings_dict) crawler.spider = crawler._create_spider() + crawler.engine = MagicMock(download_async=_mocked_download_func) self.pipeline = FilesPipeline.from_crawler(crawler) - self.pipeline.download_func = _mocked_download_func self.pipeline.open_spider() def teardown_method(self): @@ -160,8 +162,8 @@ class TestFilesPipeline: fullpath = Path(self.tempdir, "some", "image", "key.jpg") assert self.pipeline.store._get_filesystem_path(path) == fullpath - @inlineCallbacks - def test_file_not_expired(self): + @deferred_f_from_coro_f + async def test_file_not_expired(self): item_url = "http://example.com/file.pdf" item = _create_item_with_files(item_url) patchers = [ @@ -180,15 +182,15 @@ class TestFilesPipeline: for p in patchers: p.start() - result = yield self.pipeline.process_item(item) + result = await self.pipeline.process_item(item) assert result["files"][0]["checksum"] == "abc" assert result["files"][0]["status"] == "uptodate" for p in patchers: p.stop() - @inlineCallbacks - def test_file_expired(self): + @deferred_f_from_coro_f + async def test_file_expired(self): item_url = "http://example.com/file2.pdf" item = _create_item_with_files(item_url) patchers = [ @@ -211,15 +213,15 @@ class TestFilesPipeline: for p in patchers: p.start() - result = yield self.pipeline.process_item(item) + result = await self.pipeline.process_item(item) assert result["files"][0]["checksum"] != "abc" assert result["files"][0]["status"] == "downloaded" for p in patchers: p.stop() - @inlineCallbacks - def test_file_cached(self): + @deferred_f_from_coro_f + async def test_file_cached(self): item_url = "http://example.com/file3.pdf" item = _create_item_with_files(item_url) patchers = [ @@ -242,7 +244,7 @@ class TestFilesPipeline: for p in patchers: p.start() - result = yield self.pipeline.process_item(item) + result = await self.pipeline.process_item(item) assert result["files"][0]["checksum"] != "abc" assert result["files"][0]["status"] == "cached" diff --git a/tests/test_pipeline_media.py b/tests/test_pipeline_media.py index 9f174cc5f..05ad43d16 100644 --- a/tests/test_pipeline_media.py +++ b/tests/test_pipeline_media.py @@ -1,10 +1,10 @@ from __future__ import annotations import warnings +from unittest.mock import MagicMock import pytest from testfixtures import LogCapture -from twisted.internet.defer import Deferred, inlineCallbacks from twisted.python.failure import Failure from scrapy import signals @@ -13,17 +13,21 @@ from scrapy.http import Request, Response from scrapy.http.request import NO_CALLBACK from scrapy.pipelines.files import FileException from scrapy.pipelines.media import MediaPipeline -from scrapy.utils.asyncio import call_later +from scrapy.utils.defer import _defer_sleep_async, deferred_f_from_coro_f from scrapy.utils.log import failure_to_exc_info from scrapy.utils.signal import disconnect_all from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler -def _mocked_download_func(request, info): +async def _mocked_download_func(request): assert request.callback is NO_CALLBACK response = request.meta.get("response") - return response() if callable(response) else response + if callable(response): + response = await response() + if isinstance(response, Exception): + raise response + return response class UserDefinedPipeline(MediaPipeline): @@ -50,8 +54,8 @@ class TestBaseMediaPipeline: def setup_method(self): crawler = get_crawler(DefaultSpider, self.settings) crawler.spider = crawler._create_spider() + crawler.engine = MagicMock(download_async=_mocked_download_func) self.pipe = self.pipeline_class.from_crawler(crawler) - self.pipe.download_func = _mocked_download_func self.pipe.open_spider() self.info = self.pipe.spiderinfo self.fingerprint = crawler.request_fingerprinter.fingerprint @@ -158,10 +162,10 @@ class TestBaseMediaPipeline: assert new_item is item assert len(log.records) == 0 - @inlineCallbacks - def test_default_process_item(self): + @deferred_f_from_coro_f + async def test_default_process_item(self): item = {"name": "name"} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item @@ -206,8 +210,8 @@ class TestMediaPipeline(TestBaseMediaPipeline): self.pipe._mockcalled.append("request_errback") return result - @inlineCallbacks - def test_result_succeed(self): + @deferred_f_from_coro_f + async def test_result_succeed(self): rsp = Response("http://url1") req = Request( "http://url1", @@ -215,7 +219,7 @@ class TestMediaPipeline(TestBaseMediaPipeline): errback=self._errback, ) item = {"requests": req} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item["results"] == [(True, {})] assert self.pipe._mockcalled == [ "get_media_requests", @@ -224,18 +228,17 @@ class TestMediaPipeline(TestBaseMediaPipeline): "item_completed", ] - @inlineCallbacks - def test_result_failure(self): + @deferred_f_from_coro_f + async def test_result_failure(self): self.pipe.LOG_FAILED_RESULTS = False exc = Exception("foo") - fail = Failure(exc) req = Request( "http://url1", - meta={"response": fail}, + meta={"response": exc}, errback=self._errback, ) item = {"requests": req} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert len(new_item["results"]) == 1 assert new_item["results"][0][0] is False assert isinstance(new_item["results"][0][1], Failure) @@ -248,16 +251,15 @@ class TestMediaPipeline(TestBaseMediaPipeline): "item_completed", ] - @inlineCallbacks - def test_mix_of_success_and_failure(self): + @deferred_f_from_coro_f + async def test_mix_of_success_and_failure(self): self.pipe.LOG_FAILED_RESULTS = False rsp1 = Response("http://url1") req1 = Request("http://url1", meta={"response": rsp1}) exc = Exception("foo") - fail = Failure(exc) - req2 = Request("http://url2", meta={"response": fail}) + req2 = Request("http://url2", meta={"response": exc}) item = {"requests": [req1, req2]} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert len(new_item["results"]) == 2 assert new_item["results"][0] == (True, {}) assert new_item["results"][1][0] is False @@ -275,12 +277,12 @@ class TestMediaPipeline(TestBaseMediaPipeline): assert m.count("media_downloaded") == 1 assert m.count("media_failed") == 1 - @inlineCallbacks - def test_get_media_requests(self): + @deferred_f_from_coro_f + async def test_get_media_requests(self): # returns single Request (without callback) req = Request("http://url") item = {"requests": req} # pass a single item - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item assert self.fingerprint(req) in self.info.downloaded @@ -288,17 +290,17 @@ class TestMediaPipeline(TestBaseMediaPipeline): req1 = Request("http://url1") req2 = Request("http://url2") item = {"requests": iter([req1, req2])} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item assert self.fingerprint(req1) in self.info.downloaded assert self.fingerprint(req2) in self.info.downloaded - @inlineCallbacks - def test_results_are_cached_across_multiple_items(self): + @deferred_f_from_coro_f + async def test_results_are_cached_across_multiple_items(self): rsp1 = Response("http://url1") req1 = Request("http://url1", meta={"response": rsp1}) item = {"requests": req1} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item assert new_item["results"] == [(True, {})] @@ -307,25 +309,25 @@ class TestMediaPipeline(TestBaseMediaPipeline): req1.url, meta={"response": Response("http://donot.download.me")} ) item = {"requests": req2} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item assert self.fingerprint(req1) == self.fingerprint(req2) assert new_item["results"] == [(True, {})] - @inlineCallbacks - def test_results_are_cached_for_requests_of_single_item(self): + @deferred_f_from_coro_f + async def test_results_are_cached_for_requests_of_single_item(self): rsp1 = Response("http://url1") req1 = Request("http://url1", meta={"response": rsp1}) req2 = Request( req1.url, meta={"response": Response("http://donot.download.me")} ) item = {"requests": [req1, req2]} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item is item assert new_item["results"] == [(True, {}), (True, {})] - @inlineCallbacks - def test_wait_if_request_is_downloading(self): + @deferred_f_from_coro_f + async def test_wait_if_request_is_downloading(self): def _check_downloading(response): fp = self.fingerprint(req1) assert fp in self.info.downloading @@ -336,25 +338,24 @@ class TestMediaPipeline(TestBaseMediaPipeline): rsp1 = Response("http://url") - def rsp1_func(): - dfd = Deferred().addCallback(_check_downloading) - call_later(0.1, dfd.callback, rsp1) - return dfd + async def rsp1_func(): + await _defer_sleep_async() + _check_downloading(rsp1) - def rsp2_func(): + async def rsp2_func(): pytest.fail("it must cache rsp1 result and must not try to redownload") req1 = Request("http://url", meta={"response": rsp1_func}) req2 = Request(req1.url, meta={"response": rsp2_func}) item = {"requests": [req1, req2]} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item["results"] == [(True, {}), (True, {})] - @inlineCallbacks - def test_use_media_to_download_result(self): + @deferred_f_from_coro_f + async def test_use_media_to_download_result(self): req = Request("http://url", meta={"result": "ITSME"}) item = {"requests": req} - new_item = yield self.pipe.process_item(item) + new_item = await self.pipe.process_item(item) assert new_item["results"] == [(True, "ITSME")] assert self.pipe._mockcalled == [ "get_media_requests", @@ -469,45 +470,29 @@ class MediaFailedFailurePipeline(MockedMediaPipeline): return failure # deprecated -class TestMediaFailedFailure: +class TestMediaFailedFailure(TestBaseMediaPipeline): """Test that media_failed() can return a failure instead of raising.""" pipeline_class = MediaFailedFailurePipeline - settings = None - - def setup_method(self): - crawler = get_crawler(DefaultSpider, self.settings) - crawler.spider = crawler._create_spider() - self.pipe = self.pipeline_class.from_crawler(crawler) - self.pipe.download_func = _mocked_download_func - self.pipe.open_spider() - self.info = self.pipe.spiderinfo - self.fingerprint = crawler.request_fingerprinter.fingerprint - - def teardown_method(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): + @deferred_f_from_coro_f + async def test_result_failure(self): self.pipe.LOG_FAILED_RESULTS = False exc = Exception("foo") - fail = Failure(exc) req = Request( "http://url1", - meta={"response": fail}, + meta={"response": exc}, 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) + new_item = await self.pipe.process_item(item) assert len(new_item["results"]) == 1 assert new_item["results"][0][0] is False assert isinstance(new_item["results"][0][1], Failure)