Merge f71ac78f66 into e28e56aa61

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 <http-compression>` 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

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

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:

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)
 
 

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
 

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

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

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,62 @@ 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""
+
+    @coroutine_test
+    async def test_body_file_maxsize(
+        self, mockserver: MockServer, tmp_path: Path
+    ) -> None:
+        body_file = tmp_path / "body"
+        request = Request(
+            mockserver.url("/largechunkedfile", is_secure=self.is_secure),
+            meta={"body_file": str(body_file)},
+        )
+        async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler:
+            with pytest.raises(DownloadCancelledError):
+                await download_handler.download_request(request)
+        assert body_file.stat().st_size > 1_500
+
 
 class TestHttpsBase(TestHttpBase):
     is_secure = True