diff --git a/docs/topics/request-response.rst b/docs/topics/request-response.rst index 1d97e39b6..97bac4e90 100644 --- a/docs/topics/request-response.rst +++ b/docs/topics/request-response.rst @@ -827,6 +827,7 @@ Those are: * :reqmeta:`allow_offsite` * :reqmeta:`autothrottle_dont_adjust_delay` * :reqmeta:`bindaddress` +* :reqmeta:`body_file` * :reqmeta:`cookiejar` * :reqmeta:`dont_cache` * :reqmeta:`dont_merge_cookies` @@ -895,6 +896,36 @@ This meta key is not supported by :class:`~scrapy.core.downloader.handlers._httpx.HttpxDownloadHandler`, but the :setting:`DOWNLOAD_BIND_ADDRESS` is supported by it. +.. reqmeta:: body_file + +body_file +--------- + +.. versionadded:: VERSION + +File where the body of the response is written as it is downloaded, leaving the +response with an empty body: + +.. code-block:: python + + Request("https://example.org/big.zip", meta={"body_file": "big.zip"}) + +Use it to download files of any size without having to fit them in memory. + +The value can be a path, which Scrapy opens for writing and closes once the +response is downloaded, or an open, writable binary file object, which Scrapy +writes to and leaves open. Prefer a path for requests that may be retried or +redirected, since each attempt then starts the file over. + +The :ref:`compression middleware ` does not ask for a +compressed response for these requests, so that what reaches the file is the +response body itself. + +If the download fails, e.g. because the response exceeds +:setting:`DOWNLOAD_MAXSIZE`, the file may hold part of the body. + +Only the built-in HTTP download handlers support this key. + .. reqmeta:: download_timeout download_timeout diff --git a/scrapy/core/downloader/handlers/_base_streaming.py b/scrapy/core/downloader/handlers/_base_streaming.py index 5a669c732..ec48210d4 100644 --- a/scrapy/core/downloader/handlers/_base_streaming.py +++ b/scrapy/core/downloader/handlers/_base_streaming.py @@ -4,7 +4,7 @@ import base64 import logging import time from abc import ABC, abstractmethod -from io import BytesIO +from contextlib import closing from typing import TYPE_CHECKING, Any, ClassVar, Generic, NoReturn, TypedDict, TypeVar from urllib.parse import quote, urlsplit @@ -15,6 +15,7 @@ from scrapy.exceptions import ( ResponseDataLossError, ) from scrapy.utils._download_handlers import ( + _BodySink, check_stop_download, get_dataloss_msg, get_maxsize_msg, @@ -190,56 +191,56 @@ class BaseStreamingDownloadHandler(BaseHttpDownloadHandler, ABC, Generic[_Respon stop_download=stop_download, ) - response_body = BytesIO() bytes_received = 0 - try: - async for chunk in self._iter_body_chunks(response): - response_body.write(chunk) - bytes_received += len(chunk) + with closing(_BodySink(request)) as response_body: + try: + async for chunk in self._iter_body_chunks(response): + response_body.write(chunk) + bytes_received += len(chunk) - if stop_download := check_stop_download( - signals.bytes_received, self.crawler, request, data=chunk - ): + if stop_download := check_stop_download( + signals.bytes_received, self.crawler, request, data=chunk + ): + return make_response( + **make_response_base_args, + body=response_body.getvalue(), + stop_download=stop_download, + ) + + if maxsize and bytes_received > maxsize: + response_body.truncate() + self._cancel_maxsize( + bytes_received, maxsize, request, expected=False + ) + + if warnsize and bytes_received > warnsize and not reached_warnsize: + reached_warnsize = True + logger.warning( + get_warnsize_msg( + bytes_received, warnsize, request, expected=False + ) + ) + except Exception as e: + if not self._is_dataloss_exception(e): + raise + fail_on_dataloss: bool = request.meta.get( + "download_fail_on_dataloss", self._fail_on_dataloss + ) + if not fail_on_dataloss: return make_response( **make_response_base_args, body=response_body.getvalue(), - stop_download=stop_download, + flags=["dataloss"], ) + if not self._fail_on_dataloss_warned: + logger.warning(get_dataloss_msg(request.url)) + self._fail_on_dataloss_warned = True + raise ResponseDataLossError(str(e)) from e - if maxsize and bytes_received > maxsize: - response_body.truncate(0) - self._cancel_maxsize( - bytes_received, maxsize, request, expected=False - ) - - if warnsize and bytes_received > warnsize and not reached_warnsize: - reached_warnsize = True - logger.warning( - get_warnsize_msg( - bytes_received, warnsize, request, expected=False - ) - ) - except Exception as e: - if not self._is_dataloss_exception(e): - raise - fail_on_dataloss: bool = request.meta.get( - "download_fail_on_dataloss", self._fail_on_dataloss + return make_response( + **make_response_base_args, + body=response_body.getvalue(), ) - if not fail_on_dataloss: - return make_response( - **make_response_base_args, - body=response_body.getvalue(), - flags=["dataloss"], - ) - if not self._fail_on_dataloss_warned: - logger.warning(get_dataloss_msg(request.url)) - self._fail_on_dataloss_warned = True - raise ResponseDataLossError(str(e)) from e - - return make_response( - **make_response_base_args, - body=response_body.getvalue(), - ) @staticmethod def _request_headers(request: Request) -> Headers: diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index c288ae792..dba9e3cd8 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -7,7 +7,6 @@ import logging import re from contextlib import suppress from functools import partial -from io import BytesIO from time import monotonic from typing import TYPE_CHECKING, Any, TypedDict, TypeVar, cast from urllib.parse import urldefrag, urlparse @@ -48,6 +47,7 @@ from scrapy.exceptions import ( ) from scrapy.http import Headers, Response from scrapy.utils._download_handlers import ( + _BodySink, check_stop_download, get_dataloss_msg, get_maxsize_msg, @@ -532,6 +532,9 @@ class _ScrapyAgent: # deliverBody hangs for responses without body if cast("int", txresponse.length) == 0: + # No reader is created below, so the body file, if any, is created + # here, empty, as it would be for any other response. + _BodySink(request).close() return { "txresponse": txresponse, } @@ -641,7 +644,7 @@ class _ResponseReader(Protocol): self._finished: Deferred[_ResultT] = finished self._txresponse: TxResponse = txresponse self._request: Request = request - self._bodybuf: BytesIO = BytesIO() + self._bodybuf: _BodySink = _BodySink(request) self._maxsize: int = maxsize self._warnsize: int = warnsize self._fail_on_dataloss: bool = fail_on_dataloss @@ -655,10 +658,12 @@ class _ResponseReader(Protocol): def _finish_response( self, flags: list[str] | None = None, stop_download: StopDownload | None = None ) -> None: + body = self._bodybuf.getvalue() + self._bodybuf.close() self._finished.callback( { "txresponse": self._txresponse, - "body": self._bodybuf.getvalue(), + "body": body, "flags": flags, "certificate": self._certificate, "ip_address": self._ip_address, @@ -707,8 +712,8 @@ class _ResponseReader(Protocol): self._bytes_received, self._maxsize, self._request, expected=False ) ) - # Clear buffer earlier to avoid keeping data in memory for a long time. - self._bodybuf.truncate(0) + self._bodybuf.truncate() + self._bodybuf.close() self._finished.cancel() if ( @@ -747,6 +752,7 @@ class _ResponseReader(Protocol): exc.__cause__ = reason.value reason = Failure(exc) + self._bodybuf.close() self._finished.errback(reason) diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index 4fc300d90..41ee6851f 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -3,7 +3,6 @@ from __future__ import annotations import logging from contextlib import suppress from enum import Enum -from io import BytesIO from typing import TYPE_CHECKING, Any from h2.errors import ErrorCodes @@ -17,6 +16,7 @@ from scrapy import signals from scrapy.exceptions import DownloadCancelledError, StopDownload from scrapy.http.headers import Headers from scrapy.utils._download_handlers import ( + _BodySink, check_stop_download, get_maxsize_msg, get_warnsize_msg, @@ -156,7 +156,7 @@ class Stream: self._response: dict[str, Any] = { # Data received frame by frame from the server is appended # and passed to the response Deferred when completely received. - "body": BytesIO(), + "body": _BodySink(request), # The amount of data received that counts against the # flow control window "flow_controlled_size": 0, @@ -418,10 +418,9 @@ class Stream: raise StreamClosedError(self.stream_id) # The data received so far is the body of the response built for a - # stopped download, otherwise the buffer is cleared early to avoid - # keeping data in memory for a long time + # stopped download, otherwise it is discarded if reason is not StreamCloseReason.STOP_DOWNLOAD: - self._response["body"].truncate(0) + self._response["body"].truncate() self.metadata["stream_closed_local"] = True # The remote peer may have ended the stream already, e.g. because the @@ -527,6 +526,8 @@ class Stream: ) ) + self._response["body"].close() + def _fire_response_deferred(self) -> None: """Builds response from the self._response dict and fires the response deferred callback with the diff --git a/scrapy/downloadermiddlewares/httpcompression.py b/scrapy/downloadermiddlewares/httpcompression.py index 0045ddcaa..19728a544 100644 --- a/scrapy/downloadermiddlewares/httpcompression.py +++ b/scrapy/downloadermiddlewares/httpcompression.py @@ -82,6 +82,10 @@ class HttpCompressionMiddleware: def process_request( self, request: Request, spider: Spider | None = None ) -> Request | Response | None: + if "body_file" in request.meta: + # The body of these responses is written to a file as it is + # received, so there is nothing that could decompress it later. + return None request.headers.setdefault("Accept-Encoding", b", ".join(ACCEPTED_ENCODINGS)) return None diff --git a/scrapy/utils/_download_handlers.py b/scrapy/utils/_download_handlers.py index 23b715ac6..8c4f6281b 100644 --- a/scrapy/utils/_download_handlers.py +++ b/scrapy/utils/_download_handlers.py @@ -2,9 +2,12 @@ from __future__ import annotations +import os from contextlib import contextmanager from http.cookiejar import CookieJar -from typing import TYPE_CHECKING, Any +from io import BufferedIOBase, BytesIO +from pathlib import Path +from typing import TYPE_CHECKING, Any, cast from twisted.internet.defer import CancelledError from twisted.internet.error import ConnectionRefusedError as TxConnectionRefusedError @@ -33,6 +36,8 @@ if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address from urllib.request import Request as ULRequest + from _typeshed import SizedBuffer + from scrapy import Request from scrapy.crawler import Crawler from scrapy.http import Headers, Response @@ -48,6 +53,44 @@ class NullCookieJar(CookieJar): # pragma: no cover pass +class _BodySink: + """Destination of the bytes of a response body while it is downloaded. + + The bytes are kept in memory, unless the ``body_file`` request meta key is + set, in which case they are written to that file and the response ends up + with an empty body. + """ + + def __init__(self, request: Request): + body_file = request.meta.get("body_file") + self.to_file: bool = body_file is not None + self._buffer: BufferedIOBase + if isinstance(body_file, (str, os.PathLike)): + # A file opened here is also closed here, while a file object that + # comes from the meta key already open belongs to the caller. + self._buffer = Path(body_file).open("wb") + self._own_file = True + else: + self._buffer = body_file or BytesIO() + self._own_file = False + + def write(self, data: SizedBuffer) -> None: + self._buffer.write(data) + + def truncate(self) -> None: + """Discard the bytes received so far, to avoid keeping them in memory + for a long time.""" + if not self.to_file: + self._buffer.truncate(0) + + def getvalue(self) -> bytes: + return b"" if self.to_file else cast("BytesIO", self._buffer).getvalue() + + def close(self) -> None: + if self._own_file: + self._buffer.close() + + @contextmanager def wrap_twisted_exceptions() -> Iterator[None]: """Context manager that wraps Twisted exceptions into Scrapy exceptions.""" diff --git a/tests/test_downloadermiddleware_httpcompression.py b/tests/test_downloadermiddleware_httpcompression.py index 4e0fd7c8b..fd2920a2d 100644 --- a/tests/test_downloadermiddleware_httpcompression.py +++ b/tests/test_downloadermiddleware_httpcompression.py @@ -136,6 +136,11 @@ class TestHttpCompression: self.mw.process_request(request) assert request.headers.get("Accept-Encoding") == b", ".join(ACCEPTED_ENCODINGS) + def test_process_request_body_file(self): + request = Request("https://example.com", meta={"body_file": "body"}) + self.mw.process_request(request) + assert "Accept-Encoding" not in request.headers + def test_process_response_gzip(self): response = self._getresponse("gzip") assert response.request diff --git a/tests/utils/bases/download_handlers_http.py b/tests/utils/bases/download_handlers_http.py index 9b4a38724..12b1ac8bb 100644 --- a/tests/utils/bases/download_handlers_http.py +++ b/tests/utils/bases/download_handlers_http.py @@ -55,6 +55,7 @@ from tests.utils.decorators import coroutine_test if TYPE_CHECKING: from collections.abc import AsyncGenerator, Generator + from pathlib import Path from scrapy.core.downloader.handlers import DownloadHandlerProtocol from tests.mockserver.http import MockServer @@ -865,6 +866,48 @@ class TestHttpBase(ABC): response = await download_handler.download_request(request) assert response.body == path.encode() + @coroutine_test + async def test_body_file_path(self, mockserver: MockServer, tmp_path: Path) -> None: + body_file = tmp_path / "body" + request = Request( + mockserver.url("/text", is_secure=self.is_secure), + meta={"body_file": str(body_file)}, + ) + async with self.get_dh() as download_handler: + response = await download_handler.download_request(request) + assert response.body == b"" + assert body_file.read_bytes() == b"Works" + + @coroutine_test + async def test_body_file_object( + self, mockserver: MockServer, tmp_path: Path + ) -> None: + body_file = tmp_path / "body" + with body_file.open("wb") as f: + request = Request( + mockserver.url("/text", is_secure=self.is_secure), + meta={"body_file": f}, + ) + async with self.get_dh() as download_handler: + response = await download_handler.download_request(request) + assert not f.closed + assert response.body == b"" + assert body_file.read_bytes() == b"Works" + + @coroutine_test + async def test_body_file_empty_response( + self, mockserver: MockServer, tmp_path: Path + ) -> None: + body_file = tmp_path / "body" + request = Request( + mockserver.url("/text", is_secure=self.is_secure), + method="HEAD", + meta={"body_file": str(body_file)}, + ) + async with self.get_dh() as download_handler: + await download_handler.download_request(request) + assert body_file.read_bytes() == b"" + class TestHttpsBase(TestHttpBase): is_secure = True