From 4018d25a9b968ce316bf48ab9322720163d14043 Mon Sep 17 00:00:00 2001 From: paul Date: Sat, 7 Apr 2012 23:00:21 +0200 Subject: [PATCH] Use twisted.web.client.Agent for download requests (use of HTTP/1.1) Adds http11.HttpDownloadHandler in scrapy.core.downloader.handlers --- scrapy/core/downloader/handlers/http11.py | 198 ++++++++++++++++++++++ scrapy/tests/test_downloader_handlers.py | 14 +- 2 files changed, 211 insertions(+), 1 deletion(-) create mode 100644 scrapy/core/downloader/handlers/http11.py diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py new file mode 100644 index 000000000..daf1b700a --- /dev/null +++ b/scrapy/core/downloader/handlers/http11.py @@ -0,0 +1,198 @@ +"""Download handlers for http and https schemes""" + +from twisted.internet import reactor + +from scrapy.exceptions import NotSupported +from scrapy.utils.misc import load_object +from scrapy.conf import settings +from scrapy.utils.httpobj import urlparse_cached +from scrapy import optional_features + +from scrapy.utils.misc import load_object +from scrapy.conf import settings +from scrapy.http import Headers as ScrapyHeaders +from scrapy.utils.httpobj import urlparse_cached +from scrapy.responsetypes import responsetypes + + +from twisted.internet import defer, reactor, protocol +from twisted.web.client import Agent, ProxyAgent, ResponseDone, WebClientContextFactory + +from twisted.web.http_headers import Headers +from twisted.web._newclient import Response + +from twisted.web.http import PotentialDataLoss +from twisted.web._newclient import ResponseFailed +from twisted.web.iweb import IBodyProducer +from twisted.internet.error import TimeoutError + +from twisted.internet.endpoints import TCP4ClientEndpoint + +from time import time +from urlparse import urldefrag + +from zope.interface import implements + +from urlparse import urlparse, urlunparse, urldefrag + +def _parsed_url_args(parsed): + path = urlunparse(('', '', parsed.path or '/', parsed.params, parsed.query, '')) + host = parsed.hostname + port = parsed.port + scheme = parsed.scheme + netloc = parsed.netloc + if port is None: + port = 443 if scheme == 'https' else 80 + return scheme, netloc, host, port, path + +def _parse(url): + url = url.strip() + parsed = urlparse(url) + return _parsed_url_args(parsed) + + +class ScrapyAgent(object): + def __init__(self, reactor, contextFactory=WebClientContextFactory(), + connectTimeout=180, bindAddress=None): + self._reactor = reactor + self._agent = Agent(self._reactor, + contextFactory=contextFactory, + connectTimeout=connectTimeout, + bindAddress=bindAddress) + + def bindRequest(self, request): + self._scrapyrequest = request + + proxy = self._scrapyrequest.meta.get('proxy') + if proxy is not None and proxy != '': + scheme, _, host, port, _ = _parse(proxy) + endpoint = TCP4ClientEndpoint(self._reactor, host, port) + self._agent = ProxyAgent(endpoint) + self._agent._proxyEndpoint._timeout = request.meta.get('download_timeout') or self._agent._proxyEndpoint._timeout + else: + self._agent._connectTimeout = request.meta.get('download_timeout') or self._agent._connectTimeout + + def launch(self): + self._scrapyrequest._tw_start_time = time() + d = self._agent.request( + self._scrapyrequest.method, + urldefrag(self._scrapyrequest.url)[0], + Headers(self._scrapyrequest.headers), + ScrapyAgentRequestBodyProducer(self._scrapyrequest.body) if (self._scrapyrequest.body is not None) else None) + return d + + +class ScrapyAgentRequestBodyProducer(object): + implements(IBodyProducer) + + def __init__(self, body): + self.body = body + self.length = len(body) + + def startProducing(self, consumer): + consumer.write(self.body) + return defer.succeed(None) + + def pauseProducing(self): + pass + + def stopProducing(self): + pass + + +from cStringIO import StringIO +class ScrapyAgentResponseBodyReader(protocol.Protocol): + + def __init__(self, finished, response, scrapyRequest, debug=0): + + # finished is the deferred that will be fired + self._finished = finished + + self.debug = debug + + self.status = int(response.code) + self.resp_headers = list(response.headers.getAllRawHeaders()) + + self._scrapyrequest = scrapyRequest + self._scrapyrequest._tw_headers_time = time() + self._scrapyrequest.meta['download_latency'] = self._scrapyrequest._tw_headers_time - self._scrapyrequest._tw_start_time + + # body + self.bodyBuffer = StringIO() + + def dataReceived(self, bodyBytes): + if self.debug > 2: + print "dataReceived", len(bodyBytes), [bodyBytes[0:min(len(bodyBytes), 16)]] + self.bodyBuffer.write(bodyBytes) + + def connectionLost(self, reason): + + if reason.check(PotentialDataLoss): + if self.debug > 2: + print "PotentialDataLoss" + + elif reason.check(ResponseFailed): + if self.debug > 2: + print "ResponseFailed" + print reason.getErrorMessage() + reason.getBriefTraceback() + + elif reason.check(ResponseDone): + if self.debug > 2: + print "ReponseDone" + print "connection lost: %s" % reason + print 'Finished receiving body:', reason.getErrorMessage() + + # fire the deferred with Scrapy Response object + self._finished.callback(self._build_response()) + + + def _build_response(self): + headers = ScrapyHeaders(self.resp_headers) + respcls = responsetypes.from_args(headers=headers, url=urldefrag(self._scrapyrequest.url)[0]) + return respcls( + url=urldefrag(self._scrapyrequest.url)[0], + status=self.status, + headers=headers, + body=self.bodyBuffer.getvalue()) + + +class HttpDownloadHandler(object): + + def __init__(self, httpclientfactory=None): + self.debug = False + + + def download_request(self, request, spider): + """Return a deferred for the HTTP download""" + + agent = ScrapyAgent(reactor) + agent.bindRequest(request) + + d = agent.launch() + d.addCallback(self._agent_callback, request) + d.addErrback(self._agent_errback, request) + + return d + + + def _agent_callback(self, response, request): + finished = defer.Deferred() + reader = ScrapyAgentResponseBodyReader(finished, response, request, debug = 0) + response.deliverBody(reader) + + if request.method != 'HEAD': + return finished + else: + return reader._build_response() + + + def _agent_errback(self, failure, request): + if self.debug: + print "HttpDownloadHandler: errback called!" + print failure.getErrorMessage() + failure.getBriefTraceback() + failure.printTraceback() + + if failure.check(TimeoutError): + raise defer.TimeoutError diff --git a/scrapy/tests/test_downloader_handlers.py b/scrapy/tests/test_downloader_handlers.py index adf2d2433..9b7720ab1 100644 --- a/scrapy/tests/test_downloader_handlers.py +++ b/scrapy/tests/test_downloader_handlers.py @@ -12,6 +12,7 @@ from w3lib.url import path_to_file_uri from scrapy.core.downloader.handlers.file import FileDownloadHandler from scrapy.core.downloader.handlers.http import HttpDownloadHandler +from scrapy.core.downloader.handlers.http11 import HttpDownloadHandler as Http11DownloadHandler from scrapy.core.downloader.handlers.s3 import S3DownloadHandler from scrapy.spider import BaseSpider from scrapy.http import Request @@ -132,6 +133,11 @@ class HttpTestCase(unittest.TestCase): return d +class Http11TestCase(HttpTestCase): + def setUp(self): + HttpTestCase.setUp(self) + self.download_request = Http11DownloadHandler().download_request + class UriResource(resource.Resource): """Return the full uri that was requested""" @@ -177,6 +183,12 @@ class HttpProxyTestCase(unittest.TestCase): return self.download_request(request, BaseSpider('foo')).addCallback(_test) +class Http11ProxyTestCase(HttpProxyTestCase): + def setUp(self): + HttpProxyTestCase.setUp(self) + self.download_request = Http11DownloadHandler().download_request + + class HttpDownloadHandlerMock(object): def __init__(self, settings): pass @@ -240,7 +252,7 @@ class S3TestCase(unittest.TestCase): 'AWS 0PN5J17HBGZHT7JJ3X82:thdUi9VAkzhkniLj96JIrOPGi0g=') def test_request_signing5(self): - # deletes an object from the 'johnsmith' bucket using the + # deletes an object from the 'johnsmith' bucket using the # path-style and Date alternative. req = Request('s3://johnsmith/photos/puppy.jpg', \ method='DELETE', headers={