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

Await persist_file() in media pipelines. (#7182)

This commit is contained in:
Andrey Rakhmatullin 2026-03-24 14:52:56 +05:00 committed by GitHub
parent 9f4651151d
commit d42b23d78a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 123 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
docs/news.rst
View File

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

32
scrapy/pipelines/files.py
View File

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

21
scrapy/pipelines/images.py
View File

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

8
scrapy/pipelines/media.py
View File

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

75
tests/test_pipeline_files.py
View File

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

15
tests/test_pipeline_media.py
View File

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