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

Refactor HTTP download handlers (#7238)

This commit is contained in:
Andrey Rakhmatullin 2026-02-04 00:13:30 +05:00 committed by GitHub
parent 11977afba5
commit 2e53d90e4c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 248 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
scrapy/core/downloader/handlers/http11.py
View File

@ -31,7 +31,6 @@ from zope.interface import implementer
from scrapy import Request, signals
from scrapy.core.downloader.contextfactory import load_context_factory_from_settings
from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.exceptions import (
DownloadCancelledError,
DownloadTimeoutError,
@ -39,8 +38,15 @@ from scrapy.exceptions import (
StopDownload,
)
from scrapy.http import Headers, Response
from scrapy.responsetypes import responsetypes
from scrapy.utils._download_handlers import wrap_twisted_exceptions
from scrapy.utils._download_handlers import (
BaseHttpDownloadHandler,
check_stop_download,
get_dataloss_msg,
get_maxsize_msg,
get_warnsize_msg,
make_response,
wrap_twisted_exceptions,
)
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.deprecate import warn_on_deprecated_spider_attribute
from scrapy.utils.httpobj import urlparse_cached
@ -64,14 +70,14 @@ _T = TypeVar("_T")
class _ResultT(TypedDict):
txresponse: TxResponse
body: bytes
flags: list[str] | None
certificate: ssl.Certificate | None
ip_address: ipaddress.IPv4Address | ipaddress.IPv6Address | None
failure: NotRequired[Failure | None]
body: NotRequired[bytes]
flags: NotRequired[list[str] | None]
certificate: NotRequired[ssl.Certificate | None]
ip_address: NotRequired[ipaddress.IPv4Address | ipaddress.IPv6Address | None]
stop_download: NotRequired[StopDownload | None]
class HTTP11DownloadHandler(BaseDownloadHandler):
class HTTP11DownloadHandler(BaseHttpDownloadHandler):
def __init__(self, crawler: Crawler):
super().__init__(crawler)
self._crawler = crawler
@ -87,13 +93,7 @@ class HTTP11DownloadHandler(BaseDownloadHandler):
self._contextFactory: IPolicyForHTTPS = load_context_factory_from_settings(
crawler.settings, crawler
)
self._default_maxsize: int = crawler.settings.getint("DOWNLOAD_MAXSIZE")
self._default_warnsize: int = crawler.settings.getint("DOWNLOAD_WARNSIZE")
self._fail_on_dataloss: bool = crawler.settings.getbool(
"DOWNLOAD_FAIL_ON_DATALOSS"
)
self._disconnect_timeout: int = 1
self._fail_on_dataloss_warned: bool = False
async def download_request(self, request: Request) -> Response:
"""Return a deferred for the HTTP download"""
@ -121,12 +121,7 @@ class HTTP11DownloadHandler(BaseDownloadHandler):
return await maybe_deferred_to_future(agent.download_request(request))
except ResponseDataLossError:
if not self._fail_on_dataloss_warned:
logger.warning(
"Got data loss in %s. If you want to process broken "
"responses set the setting DOWNLOAD_FAIL_ON_DATALOSS = False"
" -- This message won't be shown in further requests",
request.url,
)
logger.warning(get_dataloss_msg(request.url))
self._fail_on_dataloss_warned = True
raise
@ -465,7 +460,7 @@ class ScrapyAgent:
d.addCallback(self._cb_latency, request, start_time)
# response body is ready to be consumed
d2: Deferred[_ResultT] = d.addCallback(self._cb_bodyready, request)
d3: Deferred[Response] = d2.addCallback(self._cb_bodydone, request, url)
d3: Deferred[Response] = d2.addCallback(self._cb_bodydone, url)
# check download timeout
self._timeout_cl = reactor.callLater(timeout, d3.cancel)
d3.addBoth(self._cb_timeout, request, url, timeout)
@ -497,68 +492,48 @@ class ScrapyAgent:
def _cb_bodyready(
self, txresponse: TxResponse, request: Request
) -> _ResultT | Deferred[_ResultT]:
headers_received_result = self._crawler.signals.send_catch_log(
signal=signals.headers_received,
if stop_download := check_stop_download(
signals.headers_received,
self._crawler,
request,
headers=self._headers_from_twisted_response(txresponse),
body_length=txresponse.length,
request=request,
spider=self._crawler.spider,
)
for handler, result in headers_received_result:
if isinstance(result, Failure) and isinstance(result.value, StopDownload):
logger.debug(
"Download stopped for %(request)s from signal handler %(handler)s",
{"request": request, "handler": handler.__qualname__},
)
txresponse._transport.stopProducing()
txresponse._transport.loseConnection()
return {
"txresponse": txresponse,
"body": b"",
"flags": ["download_stopped"],
"certificate": None,
"ip_address": None,
"failure": result if result.value.fail else None,
}
):
txresponse._transport.stopProducing()
txresponse._transport.loseConnection()
return {
"txresponse": txresponse,
"stop_download": stop_download,
}
# deliverBody hangs for responses without body
if txresponse.length == 0:
return {
"txresponse": txresponse,
"body": b"",
"flags": None,
"certificate": None,
"ip_address": None,
}
maxsize = request.meta.get("download_maxsize", self._maxsize)
warnsize = request.meta.get("download_warnsize", self._warnsize)
expected_size = txresponse.length if txresponse.length != UNKNOWN_LENGTH else -1
expected_size = (
cast("int", txresponse.length)
if txresponse.length != UNKNOWN_LENGTH
else -1
)
fail_on_dataloss = request.meta.get(
"download_fail_on_dataloss", self._fail_on_dataloss
)
if maxsize and expected_size > maxsize:
warning_msg = (
"Cancelling download of %(url)s: expected response "
"size (%(size)s) larger than download max size (%(maxsize)s)."
warning_msg = get_maxsize_msg(
expected_size, maxsize, request, expected=True
)
warning_args = {
"url": request.url,
"size": expected_size,
"maxsize": maxsize,
}
logger.warning(warning_msg, warning_args)
logger.warning(warning_msg)
txresponse._transport.loseConnection()
raise DownloadCancelledError(warning_msg % warning_args)
raise DownloadCancelledError(warning_msg)
if warnsize and expected_size > warnsize:
logger.warning(
"Expected response size (%(size)s) larger than "
"download warn size (%(warnsize)s) in request %(request)s.",
{"size": expected_size, "warnsize": warnsize, "request": request},
get_warnsize_msg(expected_size, warnsize, request, expected=True)
)
def _cancel(_: Any) -> None:
@ -583,31 +558,24 @@ class ScrapyAgent:
return d
def _cb_bodydone(
self, result: _ResultT, request: Request, url: str
) -> Response | Failure:
def _cb_bodydone(self, result: _ResultT, url: str) -> Response:
headers = self._headers_from_twisted_response(result["txresponse"])
respcls = responsetypes.from_args(headers=headers, url=url, body=result["body"])
try:
version = result["txresponse"].version
protocol = f"{to_unicode(version[0])}/{version[1]}.{version[2]}"
except (AttributeError, TypeError, IndexError):
protocol = None
response = respcls(
return make_response(
url=url,
status=int(result["txresponse"].code),
headers=headers,
body=result["body"],
flags=result["flags"],
certificate=result["certificate"],
ip_address=result["ip_address"],
body=result.get("body", b""),
flags=result.get("flags"),
certificate=result.get("certificate"),
ip_address=result.get("ip_address"),
protocol=protocol,
stop_download=result.get("stop_download"),
)
if result.get("failure"):
assert result["failure"]
result["failure"].value.response = response
return result["failure"]
return response
@implementer(IBodyProducer)
@ -652,7 +620,7 @@ class _ResponseReader(Protocol):
self._crawler: Crawler = crawler
def _finish_response(
self, flags: list[str] | None = None, failure: Failure | None = None
self, flags: list[str] | None = None, stop_download: StopDownload | None = None
) -> None:
self._finished.callback(
{
@ -661,7 +629,7 @@ class _ResponseReader(Protocol):
"flags": flags,
"certificate": self._certificate,
"ip_address": self._ip_address,
"failure": failure,
"stop_download": stop_download,
}
)
@ -687,32 +655,18 @@ class _ResponseReader(Protocol):
self._bodybuf.write(bodyBytes)
self._bytes_received += len(bodyBytes)
bytes_received_result = self._crawler.signals.send_catch_log(
signal=signals.bytes_received,
data=bodyBytes,
request=self._request,
spider=self._crawler.spider,
)
for handler, result in bytes_received_result:
if isinstance(result, Failure) and isinstance(result.value, StopDownload):
logger.debug(
"Download stopped for %(request)s from signal handler %(handler)s",
{"request": self._request, "handler": handler.__qualname__},
)
self.transport.stopProducing()
self.transport.loseConnection()
failure = result if result.value.fail else None
self._finish_response(flags=["download_stopped"], failure=failure)
if stop_download := check_stop_download(
signals.bytes_received, self._crawler, self._request, data=bodyBytes
):
self.transport.stopProducing()
self.transport.loseConnection()
self._finish_response(stop_download=stop_download)
if self._maxsize and self._bytes_received > self._maxsize:
logger.warning(
"Received (%(bytes)s) bytes larger than download "
"max size (%(maxsize)s) in request %(request)s.",
{
"bytes": self._bytes_received,
"maxsize": self._maxsize,
"request": self._request,
},
get_maxsize_msg(
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)
@ -725,9 +679,9 @@ class _ResponseReader(Protocol):
):
self._reached_warnsize = True
logger.warning(
"Received more bytes than download "
"warn size (%(warnsize)s) in request %(request)s.",
{"warnsize": self._warnsize, "request": self._request},
get_warnsize_msg(
self._bytes_received, self._warnsize, self._request, expected=False
)
)
def connectionLost(self, reason: Failure = connectionDone) -> None:

50
scrapy/core/http2/stream.py
View File

@ -14,7 +14,11 @@ from twisted.web.client import ResponseFailed
from scrapy.exceptions import DownloadCancelledError
from scrapy.http.headers import Headers
from scrapy.responsetypes import responsetypes
from scrapy.utils._download_handlers import (
get_maxsize_msg,
get_warnsize_msg,
make_response,
)
from scrapy.utils.httpobj import urlparse_cached
if TYPE_CHECKING:
@ -75,6 +79,9 @@ class StreamCloseReason(Enum):
# As a result sending this request will the end the connection
INVALID_HOSTNAME = 7
# Actual response body size is more than allowed limit
MAXSIZE_EXCEEDED_ACTUAL = 8
class Stream:
"""Represents a single HTTP/2 Stream.
@ -334,14 +341,16 @@ class Stream:
self._download_maxsize
and self._response["flow_controlled_size"] > self._download_maxsize
):
self.reset_stream(StreamCloseReason.MAXSIZE_EXCEEDED)
self.reset_stream(StreamCloseReason.MAXSIZE_EXCEEDED_ACTUAL)
return
if self._log_warnsize:
self.metadata["reached_warnsize"] = True
warning_msg = (
f"Received more ({self._response['flow_controlled_size']}) bytes than download "
f"warn size ({self._download_warnsize}) in request {self._request}"
warning_msg = get_warnsize_msg(
self._response["flow_controlled_size"],
self._download_warnsize,
self._request,
expected=False,
)
logger.warning(warning_msg)
@ -362,9 +371,8 @@ class Stream:
if self._log_warnsize:
self.metadata["reached_warnsize"] = True
warning_msg = (
f"Expected response size ({expected_size}) larger than "
f"download warn size ({self._download_warnsize}) in request {self._request}"
warning_msg = get_warnsize_msg(
expected_size, self._download_warnsize, self._request, expected=True
)
logger.warning(warning_msg)
@ -412,15 +420,20 @@ class Stream:
# As we immediately cancel the request when maxsize is exceeded while
# receiving DATA_FRAME's when we have received the headers (not
# having Content-Length)
if reason is StreamCloseReason.MAXSIZE_EXCEEDED:
if reason in {
StreamCloseReason.MAXSIZE_EXCEEDED,
StreamCloseReason.MAXSIZE_EXCEEDED_ACTUAL,
}:
expected_size = int(
self._response["headers"].get(
b"Content-Length", self._response["flow_controlled_size"]
)
)
error_msg = (
f"Cancelling download of {self._request.url}: received response "
f"size ({expected_size}) larger than download max size ({self._download_maxsize})"
error_msg = get_maxsize_msg(
expected_size,
self._download_maxsize,
self._request,
expected=reason == StreamCloseReason.MAXSIZE_EXCEEDED,
)
logger.error(error_msg)
self._deferred_response.errback(DownloadCancelledError(error_msg))
@ -475,22 +488,13 @@ class Stream:
and fires the response deferred callback with the
generated response instance"""
body = self._response["body"].getvalue()
response_cls = responsetypes.from_args(
headers=self._response["headers"],
url=self._request.url,
body=body,
)
response = response_cls(
response = make_response(
url=self._request.url,
status=int(self._response["headers"][":status"]),
headers=self._response["headers"],
body=body,
request=self._request,
body=self._response["body"].getvalue(),
certificate=self._protocol.metadata["certificate"],
ip_address=self._protocol.metadata["ip_address"],
protocol="h2",
)
self._deferred_response.callback(response)

7
scrapy/exceptions.py
View File

@ -7,7 +7,10 @@ new exceptions here without documenting them there.
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from scrapy.http import Response
# Internal
@ -49,6 +52,8 @@ class StopDownload(Exception):
should be handled by the request errback. Note that 'fail' is a keyword-only argument.
"""
response: Response | None
def __init__(self, *, fail: bool = True):
super().__init__()
self.fail = fail

106
scrapy/utils/_download_handlers.py
View File

@ -2,27 +2,53 @@
from __future__ import annotations
from abc import ABC
from contextlib import contextmanager
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from twisted.internet.defer import CancelledError
from twisted.internet.error import ConnectionRefusedError as TxConnectionRefusedError
from twisted.internet.error import DNSLookupError
from twisted.internet.error import TimeoutError as TxTimeoutError
from twisted.python.failure import Failure
from twisted.web.client import ResponseFailed
from twisted.web.error import SchemeNotSupported
from scrapy import responsetypes
from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.exceptions import (
CannotResolveHostError,
DownloadCancelledError,
DownloadConnectionRefusedError,
DownloadFailedError,
DownloadTimeoutError,
StopDownload,
UnsupportedURLSchemeError,
)
from scrapy.utils.log import logger
if TYPE_CHECKING:
from collections.abc import Iterator
from ipaddress import IPv4Address, IPv6Address
from twisted.internet.ssl import Certificate
from scrapy import Request
from scrapy.crawler import Crawler
from scrapy.http import Headers, Response
class BaseHttpDownloadHandler(BaseDownloadHandler, ABC):
"""Base class for built-in HTTP download handlers."""
def __init__(self, crawler: Crawler):
super().__init__(crawler)
self._default_maxsize: int = crawler.settings.getint("DOWNLOAD_MAXSIZE")
self._default_warnsize: int = crawler.settings.getint("DOWNLOAD_WARNSIZE")
self._fail_on_dataloss: bool = crawler.settings.getbool(
"DOWNLOAD_FAIL_ON_DATALOSS"
)
self._fail_on_dataloss_warned: bool = False
@contextmanager
@ -42,3 +68,81 @@ def wrap_twisted_exceptions() -> Iterator[None]:
raise DownloadFailedError(str(e)) from e
except TxTimeoutError as e:
raise DownloadTimeoutError(str(e)) from e
def check_stop_download(
signal: object, crawler: Crawler, request: Request, **kwargs: Any
) -> StopDownload | None:
"""Send the given signal and check if any of its handlers raised
:exc:`~scrapy.exceptions.StopDownload`.
Return the raised exception or ``None``.
"""
signal_result = crawler.signals.send_catch_log(
signal=signal,
request=request,
spider=crawler.spider,
**kwargs,
)
for handler, result in signal_result:
if isinstance(result, Failure) and isinstance(result.value, StopDownload):
logger.debug(
f"Download stopped for {request} from signal handler {handler.__qualname__}"
)
return result.value
return None
def make_response(
url: str,
status: int,
headers: Headers,
body: bytes = b"",
flags: list[str] | None = None,
certificate: Certificate | None = None,
ip_address: IPv4Address | IPv6Address | None = None,
protocol: str | None = None,
stop_download: StopDownload | None = None,
) -> Response:
respcls = responsetypes.responsetypes.from_args(headers=headers, url=url, body=body)
response = respcls(
url=url,
status=status,
headers=headers,
body=body,
flags=flags,
certificate=certificate,
ip_address=ip_address,
protocol=protocol,
)
if stop_download:
response.flags.append("download_stopped")
if stop_download.fail:
stop_download.response = response
raise stop_download
return response
def get_maxsize_msg(size: int, limit: int, request: Request, *, expected: bool) -> str:
prefix = "Expected to receive" if expected else "Received"
return (
f"{prefix} {size} bytes which is larger than download "
f"max size ({limit}) in request {request}."
)
def get_warnsize_msg(size: int, limit: int, request: Request, *, expected: bool) -> str:
prefix = "Expected to receive" if expected else "Received"
return (
f"{prefix} {size} bytes which is larger than download "
f"warn size ({limit}) in request {request}."
)
def get_dataloss_msg(url: str) -> str:
return (
f"Got data loss in {url}. If you want to process broken "
f"responses set the setting DOWNLOAD_FAIL_ON_DATALOSS = False"
f" -- This message won't be shown in further requests"
)

44
tests/test_downloader_handlers_http_base.py
View File

@ -4,6 +4,7 @@ from __future__ import annotations
import gzip
import json
import re
import sys
from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
@ -474,7 +475,9 @@ class TestHttp11Base(TestHttpBase):
assert type(response) is TextResponse # pylint: disable=unidiomatic-typecheck
@coroutine_test
async def test_download_with_maxsize(self, mockserver: MockServer) -> None:
async def test_download_with_maxsize(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
request = Request(mockserver.url("/text", is_secure=self.is_secure))
# 10 is minimal size for this request and the limit is only counted on
@ -483,9 +486,12 @@ class TestHttp11Base(TestHttpBase):
response = await download_handler.download_request(request)
assert response.body == b"Works"
caplog.clear()
msg = "Expected to receive 5 bytes which is larger than download max size (4)"
async with self.get_dh({"DOWNLOAD_MAXSIZE": 4}) as download_handler:
with pytest.raises(DownloadCancelledError):
with pytest.raises(DownloadCancelledError, match=re.escape(msg)):
await download_handler.download_request(request)
assert msg in caplog.text
@coroutine_test
async def test_download_with_maxsize_very_large_file(
@ -495,8 +501,10 @@ class TestHttp11Base(TestHttpBase):
async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler:
with pytest.raises(DownloadCancelledError):
await download_handler.download_request(request)
assert "larger than download max size" in caplog.text
assert (
"Received 2048 bytes which is larger than download max size (1500)"
in caplog.text
)
@coroutine_test
async def test_download_with_maxsize_per_req(self, mockserver: MockServer) -> None:
@ -524,6 +532,34 @@ class TestHttp11Base(TestHttpBase):
response = await download_handler.download_request(request)
assert response.body == b"Works"
@coroutine_test
async def test_download_with_warnsize(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
request = Request(mockserver.url("/text", is_secure=self.is_secure))
async with self.get_dh({"DOWNLOAD_WARNSIZE": 4}) as download_handler:
response = await download_handler.download_request(request)
assert response.body == b"Works"
assert (
"Expected to receive 5 bytes which is larger than download warn size (4)"
in caplog.text
)
@coroutine_test
async def test_download_with_warnsize_no_content_length(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
request = Request(
mockserver.url("/delay?n=0.1", is_secure=self.is_secure),
)
async with self.get_dh({"DOWNLOAD_WARNSIZE": 10}) as download_handler:
response = await download_handler.download_request(request)
assert response.body == b"Response delayed for 0.100 seconds\n"
assert (
"Received 35 bytes which is larger than download warn size (10)"
in caplog.text
)
@coroutine_test
async def test_download_chunked_content(self, mockserver: MockServer) -> None:
request = Request(mockserver.url("/chunked", is_secure=self.is_secure))

28
tests/test_http2_client_protocol.py
View File

@ -292,7 +292,6 @@ class TestHttps2ClientProtocol:
response = await make_request(client, request)
assert response.status == expected_status
assert response.body == expected_body
assert response.request == request
content_length_header = response.headers.get("Content-Length")
assert content_length_header is not None
@ -358,7 +357,6 @@ class TestHttps2ClientProtocol:
response = await make_request(client, request)
assert response.status == expected_status
assert response.request == request
content_length_header = response.headers.get("Content-Length")
assert content_length_header is not None
@ -465,23 +463,23 @@ class TestHttps2ClientProtocol:
d.cancel()
response = cast("Response", (yield d))
assert response.status == 499
assert response.request == request
@deferred_f_from_coro_f
async def test_download_maxsize_exceeded(
self, server_port: int, client: H2ClientProtocol
self,
caplog: pytest.LogCaptureFixture,
server_port: int,
client: H2ClientProtocol,
) -> None:
request = Request(
url=self.get_url(server_port, "/get-data-html-large"),
meta={"download_maxsize": 1000},
)
with pytest.raises(DownloadCancelledError) as exc_info:
with pytest.raises(
DownloadCancelledError,
match=r"Expected to receive \d+ bytes which is larger than download max size \(1000\)",
):
await make_request(client, request)
error_pattern = re.compile(
rf"Cancelling download of {request.url}: received response "
rf"size \(\d*\) larger than download max size \(1000\)"
)
assert len(re.findall(error_pattern, str(exc_info.value))) == 1
@inlineCallbacks
def test_received_dataloss_response(
@ -508,7 +506,6 @@ class TestHttps2ClientProtocol:
response = await make_request(client, request)
assert response.status == 200
assert response.body == Data.NO_CONTENT_LENGTH
assert response.request == request
assert "Content-Length" not in response.headers
async def _check_log_warnsize(
@ -522,7 +519,6 @@ class TestHttps2ClientProtocol:
with caplog.at_level("WARNING", "scrapy.core.http2.stream"):
response = await make_request(client, request)
assert response.status == 200
assert response.request == request
assert response.body == expected_body
# Check the warning is raised only once for this request
@ -540,7 +536,7 @@ class TestHttps2ClientProtocol:
meta={"download_warnsize": 1000},
)
warn_pattern = re.compile(
rf"Expected response size \(\d*\) larger than "
rf"Expected to receive \d+ bytes which is larger than "
rf"download warn size \(1000\) in request {request}"
)
@ -560,8 +556,8 @@ class TestHttps2ClientProtocol:
meta={"download_warnsize": 10},
)
warn_pattern = re.compile(
rf"Received more \(\d*\) bytes than download "
rf"warn size \(10\) in request {request}"
rf"Received \d+ bytes which is larger than "
rf"download warn size \(10\) in request {request}"
)
await self._check_log_warnsize(
@ -671,7 +667,6 @@ class TestHttps2ClientProtocol:
) -> None:
request = Request(self.get_url(server_port, "/status?n=200"))
response = await make_request(client, request)
assert response.request == request
assert isinstance(response.certificate, Certificate)
assert response.certificate.original is not None
assert response.certificate.getIssuer() == client_certificate.getIssuer()
@ -749,7 +744,6 @@ class TestHttps2ClientProtocol:
)
response = await make_request(client, request)
assert response.status == 200
assert response.request == request
response_headers = json.loads(str(response.body, "utf-8"))
assert isinstance(response_headers, dict)