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

Move itempipeline functionality outside of engine as a spidermiddleware

This commit is contained in:
Daniel Grana 2009-07-06 15:31:50 -03:00
parent 2ce43ebbec
commit 3cb18dbbbb
4 changed files with 86 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
scrapy/conf/default_settings.py
View File

@ -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,

75
scrapy/contrib/spidermiddleware/itempipeline.py Normal file
View File

@ -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)

31
scrapy/core/engine.py
View File

@ -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])",
]

4
scrapy/item/pipeline.py
View File

@ -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