From 3cb18dbbbb315195d6bd12e7461362d6610a9ccf Mon Sep 17 00:00:00 2001 From: Daniel Grana Date: Mon, 6 Jul 2009 15:31:50 -0300 Subject: [PATCH] Move itempipeline functionality outside of engine as a spidermiddleware --- scrapy/conf/default_settings.py | 4 + .../contrib/spidermiddleware/itempipeline.py | 75 +++++++++++++++++++ scrapy/core/engine.py | 31 +------- scrapy/item/pipeline.py | 4 +- 4 files changed, 86 insertions(+), 28 deletions(-) create mode 100644 scrapy/contrib/spidermiddleware/itempipeline.py diff --git a/scrapy/conf/default_settings.py b/scrapy/conf/default_settings.py index a8519a25d..1bb4bd4f1 100644 --- a/scrapy/conf/default_settings.py +++ b/scrapy/conf/default_settings.py @@ -114,6 +114,9 @@ HTTPCACHE_EXPIRATION_SECS = 0 # Item pipelines are typically set in specific commands settings ITEM_PIPELINES = [] +# max limit of items to process in parallel +ITEMPIPELINE_CONCURRENTLIMIT = 0 + LOG_ENABLED = True LOG_STDOUT = False LOGLEVEL = 'DEBUG' @@ -169,6 +172,7 @@ SPIDER_MIDDLEWARES = {} SPIDER_MIDDLEWARES_BASE = { # Engine side + 'scrapy.contrib.spidermiddleware.itempipeline.ItemPipelineMiddleware': 30, 'scrapy.contrib.spidermiddleware.httperror.HttpErrorMiddleware': 50, 'scrapy.contrib.itemsampler.ItemSamplerMiddleware': 100, 'scrapy.contrib.spidermiddleware.limit.RequestLimitMiddleware': 200, diff --git a/scrapy/contrib/spidermiddleware/itempipeline.py b/scrapy/contrib/spidermiddleware/itempipeline.py new file mode 100644 index 000000000..66a62a750 --- /dev/null +++ b/scrapy/contrib/spidermiddleware/itempipeline.py @@ -0,0 +1,75 @@ +""" +ItemPipelineMiddleware: feed item pipeline with scraped items +""" +from pydispatch import dispatcher + +from twisted.python.failure import Failure + +from scrapy.core import signals +from scrapy.core.exceptions import DontCloseDomain +from scrapy.item.pipeline import ItemPipelineManager +from scrapy.item import ScrapedItem +from scrapy.conf import settings +from scrapy import log + +class ItemPipelineMiddleware(object): + """SpiderMiddleware that sends items through a pipeline""" + + # The type of items to process by pipeline + ScrapedItem = ScrapedItem + + # The Pipeline Manager to use for processing these item + ItemPipelineManager = ItemPipelineManager + + # Maximum number of items to process in parallel by this pipeline + concurrent_limit = settings.getint('ITEMPIPELINE_CONCURRENTLIMIT', 0) + + def __init__(self): + self.pipeline = self.ItemPipelineManager() + dispatcher.connect(self.domain_opened, signal=signals.domain_opened) + dispatcher.connect(self.domain_closed, signal=signals.domain_closed) + dispatcher.connect(self.domain_idle, signal=signals.domain_idle) + + def domain_opened(self, domain): + self.pipeline.open_domain(domain) + + def domain_closed(self, domain): + self.pipeline.close_domain(domain) + + def domain_idle(self, domain): + if not self.pipeline.domain_is_idle(domain): + raise DontCloseDomain + + def process_spider_output(self, response, result, spider): + domain = spider.domain_name + info = self.pipeline.domaininfo[domain] + + for item_or_request in result: + # return to engine until pipeline frees up some slots + # TODO: this is ugly, a proper flow control mechanism should be + # added instead + while 0 < self.concurrent_limit <= len(info): + yield None + + if isinstance(item_or_request, self.ScrapedItem): + log.msg("Scraped %s in <%s>" % (item_or_request, response.request.url), \ + domain=domain) + signals.send_catch_log(signal=signals.item_scraped, sender=self.__class__, \ + item=item_or_request, spider=spider, response=response) + self.pipeline.pipe(item_or_request, spider).addBoth(self._pipeline_finished, \ + item_or_request, spider) + # yielding here breaks the loop and allows the engine to run + # other tasks, such as attending IO (very important) + yield None + else: + yield item_or_request + + def _pipeline_finished(self, pipe_result, item, spider): + # exception can only be of DropItem type here, since other exceptions + # are caught in the Item Pipeline (item/pipeline.py) + if isinstance(pipe_result, Failure): + signals.send_catch_log(signal=signals.item_dropped, \ + sender=self.__class__, item=item, spider=spider, exception=pipe_result.value) + else: + signals.send_catch_log(signal=signals.item_passed, \ + sender=self.__class__, item=item, spider=spider, pipe_output=pipe_result) diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index bc8cbe6e3..e1050f084 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -20,8 +20,6 @@ from scrapy.core.scheduler import Scheduler from scrapy.core.downloader import Downloader from scrapy.core.exceptions import IgnoreRequest, DontCloseDomain from scrapy.http import Response, Request -from scrapy.item import ScrapedItem -from scrapy.item.pipeline import ItemPipelineManager from scrapy.spider import spiders from scrapy.spider.middleware import SpiderMiddlewareManager from scrapy.utils.misc import load_object @@ -49,8 +47,6 @@ class ExecutionEngine(object): self.downloader = downloader or Downloader() self.spidermiddleware = SpiderMiddlewareManager() self._scraping = {} - self.pipeline = ItemPipelineManager() - self.configured = True def addtask(self, function, interval, args=None, kwargs=None, now=False): @@ -140,7 +136,7 @@ class ExecutionEngine(object): self.paused = False def is_idle(self): - return self.scheduler.is_idle() and self.pipeline.is_idle() and self.downloader.is_idle() and not self._scraping + return self.scheduler.is_idle() and self.downloader.is_idle() and not self._scraping def next_domain(self): domain = self.domain_scheduler.next_domain() @@ -186,8 +182,7 @@ class ExecutionEngine(object): scraping = self._scraping.get(domain) pending = self.scheduler.domain_has_pending_requests(domain) downloading = domain in self.downloader.sites and self.downloader.sites[domain].active - haspipe = not self.pipeline.domain_is_idle(domain) - return not (pending or downloading or haspipe or scraping) + return not (pending or downloading or scraping) def domain_is_closed(self, domain): """Return True if the domain is fully closed (ie. not even in the @@ -207,25 +202,13 @@ class ExecutionEngine(object): domain = spider.domain_name def _process_response(response): - assert isinstance(response, (Response, Exception)), "Expecting Response or Exception, got %s" % type(response).__name__ + assert isinstance(response, (Response, Exception)), \ + "Expecting Response or Exception, got %s" % type(response).__name__ def cb_spidermiddleware_output(spmw_result): def cb_spider_output(output): - def cb_pipeline_output(pipe_result, item): - if isinstance(pipe_result, Failure): - # can only be a DropItem exception, since other exceptions are caught in the Item Pipeline (item/pipeline.py) - signals.send_catch_log(signal=signals.item_dropped, sender=self.__class__, item=item, spider=spider, response=response, exception=pipe_result.value) - else: - signals.send_catch_log(signal=signals.item_passed, sender=self.__class__, item=item, spider=spider, response=response, pipe_output=pipe_result) - self.next_request(domain) - if domain in self.closing: return - elif isinstance(output, ScrapedItem): - log.msg("Scraped %s in <%s>" % (output, request.url), log.INFO, domain=domain) - signals.send_catch_log(signal=signals.item_scraped, sender=self.__class__, item=output, spider=spider, response=response) - piped = self.pipeline.pipe(output, spider) - piped.addBoth(cb_pipeline_output, output) elif isinstance(output, Request): signals.send_catch_log(signal=signals.request_received, sender=self.__class__, request=output, spider=spider, response=response) self.crawl(request=output, spider=spider) @@ -330,7 +313,6 @@ class ExecutionEngine(object): self.next_request(domain) self.downloader.open_domain(domain) - self.pipeline.open_domain(domain) self._scraping[domain] = set() signals.send_catch_log(signals.domain_open, sender=self.__class__, domain=domain, spider=spider) @@ -380,7 +362,6 @@ class ExecutionEngine(object): """This function is called after the domain has been closed""" spider = spiders.fromdomain(domain) self.scheduler.close_domain(domain) - self.pipeline.close_domain(domain) del self._scraping[domain] reason = self.closing.pop(domain, 'finished') signals.send_catch_log(signal=signals.domain_closed, sender=self.__class__, domain=domain, spider=spider, reason=reason) @@ -401,8 +382,6 @@ class ExecutionEngine(object): "self.downloader.is_idle()", "len(self.downloader.sites)", "self.downloader.has_capacity()", - "self.pipeline.is_idle()", - "len(self.pipeline.domaininfo)", "len(self._scraping)", ] domain_tests = [ @@ -415,8 +394,6 @@ class ExecutionEngine(object): "len(self.downloader.sites[domain].transferring)", "self.downloader.sites[domain].closing", "self.downloader.sites[domain].lastseen", - "self.pipeline.domain_is_idle(domain)", - "len(self.pipeline.domaininfo[domain])", "len(self._scraping[domain])", ] diff --git a/scrapy/item/pipeline.py b/scrapy/item/pipeline.py index 51e949164..edb535300 100644 --- a/scrapy/item/pipeline.py +++ b/scrapy/item/pipeline.py @@ -62,7 +62,9 @@ class ItemPipelineManager(object): info.add(item) def _next_stage(item): - assert isinstance(item, ScrapedItem), 'Pipeline stages must return a ScrapedItem or raise DropItem' + assert isinstance(item, ScrapedItem), \ + 'Pipeline stages must return a ScrapedItem or raise DropItem, got %s' % type(item).__name__ + if not pipeline: return item