From 6f02a8dccb95373f22bac18c08d9fda8169dcb02 Mon Sep 17 00:00:00 2001 From: Eugenio Lacuesta Date: Wed, 29 Jan 2020 14:53:23 -0300 Subject: [PATCH] Add source parameter to bytes_received signal --- docs/topics/signals.rst | 12 ++++++++---- scrapy/core/downloader/handlers/http11.py | 18 +++++++++++++----- scrapy/core/downloader/handlers/s3.py | 1 + tests/test_downloader_handlers.py | 3 +++ tests/test_engine.py | 5 ++++- 5 files changed, 29 insertions(+), 10 deletions(-) diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst index f490911f3..3a15bf95c 100644 --- a/docs/topics/signals.rst +++ b/docs/topics/signals.rst @@ -162,11 +162,11 @@ bytes_received -------------- .. signal:: bytes_received -.. function:: bytes_received(data, request, spider) +.. function:: bytes_received(data, request, spider, source) - Sent by the HTTP 1.1 download handler when a group of bytes is - received for a specific request. This signal might be fired - multiple times for the same request. + Sent by the HTTP 1.1 and S3 download handlers when a group of bytes is + received for a specific request. This signal might be fired multiple + times for the same request, with partial data each time. This signal does not support returning deferreds from its handlers. @@ -179,6 +179,10 @@ bytes_received :param spider: the spider associated with the response :type spider: :class:`~scrapy.spiders.Spider` object + :param source: a string to identify which handler sent the signal + (current values could be "http11" or "s3") + :type source: :class:`str` object + spider_closed ------------- diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index 6f1bd9ad6..49c9eacac 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -32,8 +32,9 @@ logger = logging.getLogger(__name__) class HTTP11DownloadHandler: lazy = False - def __init__(self, settings, crawler=None): + def __init__(self, settings, crawler=None, source="http11"): self.crawler = crawler + self.source = source self._pool = HTTPConnectionPool(reactor, persistent=True) self._pool.maxPersistentPerHost = settings.getint('CONCURRENT_REQUESTS_PER_DOMAIN') self._pool._factory.noisy = False @@ -67,8 +68,8 @@ class HTTP11DownloadHandler: self._disconnect_timeout = 1 @classmethod - def from_crawler(cls, crawler): - return cls(crawler.settings, crawler) + def from_crawler(cls, crawler, **kwargs): + return cls(crawler.settings, crawler, **kwargs) def download_request(self, request, spider): """Return a deferred for the HTTP download""" @@ -79,6 +80,7 @@ class HTTP11DownloadHandler: warnsize=getattr(spider, 'download_warnsize', self._default_warnsize), fail_on_dataloss=self._fail_on_dataloss, crawler=self.crawler, + source=self.source, ) return agent.download_request(request) @@ -275,7 +277,7 @@ class ScrapyAgent: _TunnelingAgent = TunnelingAgent def __init__(self, contextFactory=None, connectTimeout=10, bindAddress=None, pool=None, - maxsize=0, warnsize=0, fail_on_dataloss=True, crawler=None): + maxsize=0, warnsize=0, fail_on_dataloss=True, crawler=None, source=None): self._contextFactory = contextFactory self._connectTimeout = connectTimeout self._bindAddress = bindAddress @@ -285,6 +287,7 @@ class ScrapyAgent: self._fail_on_dataloss = fail_on_dataloss self._txresponse = None self._crawler = crawler + self._source = source def _get_agent(self, request, timeout): bindaddress = request.meta.get('bindaddress') or self._bindAddress @@ -421,6 +424,7 @@ class ScrapyAgent: warnsize, fail_on_dataloss, self._crawler, + self._source, ) ) @@ -457,7 +461,9 @@ class _RequestBodyProducer: class _ResponseReader(protocol.Protocol): - def __init__(self, finished, txresponse, request, maxsize, warnsize, fail_on_dataloss, crawler): + def __init__( + self, finished, txresponse, request, maxsize, warnsize, fail_on_dataloss, crawler, source + ): self._finished = finished self._txresponse = txresponse self._request = request @@ -469,6 +475,7 @@ class _ResponseReader(protocol.Protocol): self._reached_warnsize = False self._bytes_received = 0 self._crawler = crawler + self._source = source def dataReceived(self, bodyBytes): # This maybe called several times after cancel was called with buffered data. @@ -483,6 +490,7 @@ class _ResponseReader(protocol.Protocol): data=bodyBytes, request=self._request, spider=self._crawler.spider, + source=self._source, ) if self._maxsize and self._bytes_received > self._maxsize: diff --git a/scrapy/core/downloader/handlers/s3.py b/scrapy/core/downloader/handlers/s3.py index 40a1fa48e..2366b6394 100644 --- a/scrapy/core/downloader/handlers/s3.py +++ b/scrapy/core/downloader/handlers/s3.py @@ -73,6 +73,7 @@ class S3DownloadHandler: objcls=httpdownloadhandler, settings=settings, crawler=crawler, + source="s3", ) self._download_http = _http_handler.download_request diff --git a/tests/test_downloader_handlers.py b/tests/test_downloader_handlers.py index 8d95d7cac..22a813647 100644 --- a/tests/test_downloader_handlers.py +++ b/tests/test_downloader_handlers.py @@ -730,6 +730,9 @@ class Http11ProxyTestCase(HttpProxyTestCase): class HttpDownloadHandlerMock: + def __init__(self, *args, **kwargs): + pass + def download_request(self, request, spider): return request diff --git a/tests/test_engine.py b/tests/test_engine.py index 3c5cc403b..c83a23b55 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -112,6 +112,7 @@ class CrawlerRun(object): self.itemerror = [] self.itemresp = [] self.bytes = defaultdict(lambda: list()) + self.bytes_source = set() self.signals_caught = {} self.spider_class = spider_class @@ -165,8 +166,9 @@ class CrawlerRun(object): def item_scraped(self, item, spider, response): self.itemresp.append((item, response)) - def bytes_received(self, data, request, spider): + def bytes_received(self, data, request, spider, source): self.bytes[request].append(data) + self.bytes_source.add(source) def request_scheduled(self, request, spider): self.reqplug.append((request, spider)) @@ -279,6 +281,7 @@ class EngineTest(unittest.TestCase): def _assert_bytes_received(self): self.assertEqual(9, len(self.run.bytes)) + self.assertEqual(self.run.bytes_source, set(["http11"])) for request, data in self.run.bytes.items(): joined_data = b"".join(data) if self.run.getpath(request.url) == "/":