From d42b23d78a4f14e53837b67cdec527c4c9c48dee Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 24 Mar 2026 14:52:56 +0500 Subject: [PATCH] Await persist_file() in media pipelines. (#7182) --- docs/news.rst | 18 +++++++++ scrapy/pipelines/files.py | 32 ++++++++++----- scrapy/pipelines/images.py | 21 +++++----- scrapy/pipelines/media.py | 8 +++- tests/test_pipeline_files.py | 75 +++++++++++++++++++++++------------- tests/test_pipeline_media.py | 15 ++++++++ 6 files changed, 123 insertions(+), 46 deletions(-) diff --git a/docs/news.rst b/docs/news.rst index cbdef67fc..6bff95518 100644 --- a/docs/news.rst +++ b/docs/news.rst @@ -6,6 +6,24 @@ Release notes Scrapy VERSION (unreleased) --------------------------- +Backward-incompatible changes +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +- In order to fix a long-standing bug with handling of asynchronous storages + the following changes were made to media pipeline classes, which can impact + some of the user code that subclasses them or calls their methods directly: + + - overrides of :meth:`scrapy.pipelines.media.MediaPipeline.media_downloaded` + and :meth:`~scrapy.pipelines.files.FilesPipeline.file_downloaded` can now + return coroutines + + - :meth:`~scrapy.pipelines.files.FilesPipeline.media_downloaded`, + :meth:`~scrapy.pipelines.files.FilesPipeline.file_downloaded` and + :meth:`~scrapy.pipelines.images.ImagesPipeline.image_downloaded` now + return coroutines + + (:issue:`2183`, :issue:`6369`, :issue:`7182`) + New features ~~~~~~~~~~~~ diff --git a/scrapy/pipelines/files.py b/scrapy/pipelines/files.py index 55e01c4ab..a0bb674d1 100644 --- a/scrapy/pipelines/files.py +++ b/scrapy/pipelines/files.py @@ -31,12 +31,14 @@ from scrapy.http.request import NO_CALLBACK from scrapy.pipelines.media import FileInfo, FileInfoOrError, MediaPipeline from scrapy.utils.boto import is_botocore_available from scrapy.utils.datatypes import CaseInsensitiveDict +from scrapy.utils.defer import ensure_awaitable from scrapy.utils.ftp import ftp_store_file from scrapy.utils.log import failure_to_exc_info from scrapy.utils.python import to_bytes from scrapy.utils.request import referer_str if TYPE_CHECKING: + from collections.abc import Awaitable from os import PathLike from twisted.python.failure import Failure @@ -591,7 +593,7 @@ class FilesPipeline(MediaPipeline): raise FileException - def media_downloaded( + async def media_downloaded( self, response: Response, request: Request, @@ -630,7 +632,9 @@ class FilesPipeline(MediaPipeline): try: path = self.file_path(request, response=response, info=info, item=item) - checksum = self.file_downloaded(response, request, info, item=item) + checksum: str = await ensure_awaitable( + self.file_downloaded(response, request, info, item=item) + ) except FileException as exc: logger.warning( "File (error): Error processing file from %(request)s " @@ -662,6 +666,21 @@ class FilesPipeline(MediaPipeline): self.crawler.stats.inc_value("file_count") self.crawler.stats.inc_value(f"file_status_count/{status}") + async def _file_downloaded( + self, + response: Response, + request: Request, + info: MediaPipeline.SpiderInfo, + *, + item: Any = None, + ) -> str: + path = self.file_path(request, response=response, info=info, item=item) + buf = BytesIO(response.body) + checksum = _md5sum(buf) + buf.seek(0) + await ensure_awaitable(self.store.persist_file(path, buf, info)) + return checksum + # Overridable Interface def get_media_requests( self, item: Any, info: MediaPipeline.SpiderInfo @@ -680,13 +699,8 @@ class FilesPipeline(MediaPipeline): info: MediaPipeline.SpiderInfo, *, item: Any = None, - ) -> str: - path = self.file_path(request, response=response, info=info, item=item) - buf = BytesIO(response.body) - checksum = _md5sum(buf) - buf.seek(0) - self.store.persist_file(path, buf, info) - return checksum + ) -> str | Awaitable[str]: + return self._file_downloaded(response, request, info, item=item) def item_completed( self, results: list[FileInfoOrError], item: Any, info: MediaPipeline.SpiderInfo diff --git a/scrapy/pipelines/images.py b/scrapy/pipelines/images.py index f5d1ad1b9..e92f33cca 100644 --- a/scrapy/pipelines/images.py +++ b/scrapy/pipelines/images.py @@ -19,6 +19,7 @@ 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.defer import ensure_awaitable from scrapy.utils.python import to_bytes if TYPE_CHECKING: @@ -115,7 +116,7 @@ class ImagesPipeline(FilesPipeline): store_uri = settings["IMAGES_STORE"] return cls(store_uri, crawler=crawler) - def file_downloaded( + async def file_downloaded( self, response: Response, request: Request, @@ -123,9 +124,9 @@ class ImagesPipeline(FilesPipeline): *, item: Any = None, ) -> str: - return self.image_downloaded(response, request, info, item=item) + return await self.image_downloaded(response, request, info, item=item) - def image_downloaded( + async def image_downloaded( self, response: Response, request: Request, @@ -139,12 +140,14 @@ class ImagesPipeline(FilesPipeline): buf.seek(0) checksum = _md5sum(buf) width, height = image.size - self.store.persist_file( - path, - buf, - info, - meta={"width": width, "height": height}, - headers={"Content-Type": "image/jpeg"}, + await ensure_awaitable( + self.store.persist_file( + path, + buf, + info, + meta={"width": width, "height": height}, + headers={"Content-Type": "image/jpeg"}, + ) ) assert checksum is not None return checksum diff --git a/scrapy/pipelines/media.py b/scrapy/pipelines/media.py index 7bc29f28f..b0ee10c12 100644 --- a/scrapy/pipelines/media.py +++ b/scrapy/pipelines/media.py @@ -31,6 +31,8 @@ from scrapy.utils.python import global_object_name if TYPE_CHECKING: # typing.Self requires Python 3.11 + from collections.abc import Awaitable + from typing_extensions import Self from scrapy import Spider @@ -209,7 +211,9 @@ class MediaPipeline(ABC): 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) + return await ensure_awaitable( + self.media_downloaded(response, request, info, item=item) + ) except Exception: failure = self.media_failed(Failure(), request, info) if isinstance(failure, Failure): @@ -282,7 +286,7 @@ class MediaPipeline(ABC): info: SpiderInfo, *, item: Any = None, - ) -> FileInfo: + ) -> FileInfo | Awaitable[FileInfo]: """Handler for success downloads""" raise NotImplementedError diff --git a/tests/test_pipeline_files.py b/tests/test_pipeline_files.py index 57736e0ea..7f59af7de 100644 --- a/tests/test_pipeline_files.py +++ b/tests/test_pipeline_files.py @@ -19,6 +19,7 @@ from urllib.parse import urlparse import attr import pytest from itemadapter import ItemAdapter +from twisted.internet.defer import Deferred from scrapy.exceptions import NotConfigured from scrapy.http import Request, Response @@ -31,6 +32,7 @@ from scrapy.pipelines.files import ( S3FilesStore, ) from scrapy.settings import Settings +from scrapy.utils.asyncio import call_later from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler from tests.mockserver.ftp import MockFTPServer @@ -78,6 +80,22 @@ def get_ftp_content_and_delete( return b"".join(ftp_data) +class DeferredFSFilesStore(FSFilesStore): + """A simple store with persist_file() returning a deferred.""" + + def persist_file(self, path, buf, info, meta=None, headers=None): + deferred = Deferred() + # short-hand super() doesn't work in nested functions + parent_persist_file = super().persist_file + + def cb(): + parent_persist_file(path, buf, info, meta=meta, headers=headers) + deferred.callback(None) + + call_later(0.5, cb) + return deferred + + class TestFilesPipeline: def setup_method(self): self.tempdir = mkdtemp() @@ -165,7 +183,7 @@ class TestFilesPipeline: async def test_file_not_expired(self): item_url = "http://example.com/file.pdf" item = _create_item_with_files(item_url) - patchers = [ + with ( mock.patch.object(FilesPipeline, "inc_stats", return_value=True), mock.patch.object( FSFilesStore, @@ -177,22 +195,16 @@ class TestFilesPipeline: "get_media_requests", return_value=[_prepare_request_object(item_url)], ), - ] - for p in patchers: - p.start() - - result = await 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() - @coroutine_test async def test_file_expired(self): item_url = "http://example.com/file2.pdf" item = _create_item_with_files(item_url) - patchers = [ + with ( mock.patch.object( FSFilesStore, "stat_file", @@ -208,22 +220,16 @@ class TestFilesPipeline: return_value=[_prepare_request_object(item_url)], ), mock.patch.object(FilesPipeline, "inc_stats", return_value=True), - ] - for p in patchers: - p.start() - - result = await 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() - @coroutine_test async def test_file_cached(self): item_url = "http://example.com/file3.pdf" item = _create_item_with_files(item_url) - patchers = [ + with ( mock.patch.object(FilesPipeline, "inc_stats", return_value=True), mock.patch.object( FSFilesStore, @@ -239,16 +245,33 @@ class TestFilesPipeline: "get_media_requests", return_value=[_prepare_request_object(item_url, flags=["cached"])], ), - ] - for p in patchers: - p.start() - - result = await self.pipeline.process_item(item) + ): + result = await self.pipeline.process_item(item) assert result["files"][0]["checksum"] != "abc" assert result["files"][0]["status"] == "cached" - for p in patchers: - p.stop() + @coroutine_test + async def test_async_store(self) -> None: + """Test that async persist_file() works and is awaited.""" + + self.pipeline.store = DeferredFSFilesStore(self.tempdir) + item_url = "http://example.com/file.pdf" + item = _create_item_with_files(item_url) + with ( + mock.patch.object(FilesPipeline, "inc_stats", return_value=True), + mock.patch.object( + FilesPipeline, + "get_media_requests", + return_value=[_prepare_request_object(item_url)], + ), + ): + result = await self.pipeline.process_item(item) + assert result["files"][0]["status"] == "downloaded" + assert result["files"][0]["checksum"] + # check that the file was written by persist_file() + path = Path(self.tempdir) / result["files"][0]["path"] + assert path.exists() + assert path.read_bytes() == b"data" def test_file_path_from_item(self): """ diff --git a/tests/test_pipeline_media.py b/tests/test_pipeline_media.py index 8229d1da1..44df0bdd4 100644 --- a/tests/test_pipeline_media.py +++ b/tests/test_pipeline_media.py @@ -204,6 +204,11 @@ class MockedMediaPipeline(UserDefinedPipeline): return item +class AsyncMediaDownloadedPipeline(MockedMediaPipeline): + async def media_downloaded(self, response, request, info, *, item=None): + return super().media_downloaded(response, request, info) + + class TestMediaPipeline(TestBaseMediaPipeline): pipeline_class = MockedMediaPipeline @@ -371,6 +376,16 @@ class TestMediaPipeline(TestBaseMediaPipeline): ) +class TestAsyncMediaDownloaded(TestMediaPipeline): + pipeline_class = AsyncMediaDownloadedPipeline + + def test_key_for_pipe(self): + assert ( + self.pipe._key_for_pipe("IMAGES", base_class_name="MediaPipeline") + == "ASYNCMEDIADOWNLOADEDPIPELINE_IMAGES" + ) + + class TestMediaPipelineAllowRedirectSettings: def _assert_request_no3xx(self, pipeline_class, settings): pipe = pipeline_class(crawler=get_crawler(None, settings))