RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor MediaPipeline.process_item() to async def (#7177)

This commit is contained in:
Andrey Rakhmatullin 2025-12-15 14:02:04 +05:00 committed by GitHub
parent 588f3d4f65
commit c49ae2115a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 148 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
docs/news.rst
View File

@ -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)

17
scrapy/pipelines/files.py
View File

@ -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:

18
scrapy/pipelines/images.py
View File

@ -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(

93
scrapy/pipelines/media.py
View File

@ -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)

17
scrapy/utils/defer.py
View File

@ -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]:

22
tests/test_pipeline_files.py
View File

@ -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"

113
tests/test_pipeline_media.py
View File

@ -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)