Add async callback support to the parse command (#5577)

scrapy/commands/parse.py

@@ -5,6 +5,8 @@ from typing import Dict
 from itemadapter import is_item, ItemAdapter
 from w3lib.url import is_url
 
+from twisted.internet.defer import maybeDeferred
+
 from scrapy.commands import BaseRunSpiderCommand
 from scrapy.http import Request
 from scrapy.utils import display
@@ -110,16 +112,19 @@ class Command(BaseRunSpiderCommand):
             if not opts.nolinks:
                 self.print_requests(colour=colour)
 
-    def run_callback(self, response, callback, cb_kwargs=None):
-        cb_kwargs = cb_kwargs or {}
+    def _get_items_and_requests(self, spider_output, opts, depth, spider, callback):
         items, requests = [], []
-
-        for x in iterate_spider_output(callback(response, **cb_kwargs)):
+        for x in spider_output:
             if is_item(x):
                 items.append(x)
             elif isinstance(x, Request):
                 requests.append(x)
-        return items, requests
+        return items, requests, opts, depth, spider, callback
+
+    def run_callback(self, response, callback, cb_kwargs=None):
+        cb_kwargs = cb_kwargs or {}
+        d = maybeDeferred(iterate_spider_output, callback(response, **cb_kwargs))
+        return d
 
     def get_callback_from_rules(self, spider, response):
         if getattr(spider, 'rules', None):
@@ -158,6 +163,25 @@ class Command(BaseRunSpiderCommand):
             logger.error('No response downloaded for: %(url)s',
                          {'url': url})
 
+    def scraped_data(self, args):
+        items, requests, opts, depth, spider, callback = args
+        if opts.pipelines:
+            itemproc = self.pcrawler.engine.scraper.itemproc
+            for item in items:
+                itemproc.process_item(item, spider)
+        self.add_items(depth, items)
+        self.add_requests(depth, requests)
+
+        scraped_data = items if opts.output else []
+        if depth < opts.depth:
+            for req in requests:
+                req.meta['_depth'] = depth + 1
+                req.meta['_callback'] = req.callback
+                req.callback = callback
+            scraped_data += requests
+
+        return scraped_data
+
     def prepare_request(self, spider, request, opts):
         def callback(response, **cb_kwargs):
             # memorize first request
@@ -191,23 +215,10 @@ class Command(BaseRunSpiderCommand):
             # parse items and requests
             depth = response.meta['_depth']
 
-            items, requests = self.run_callback(response, cb, cb_kwargs)
-            if opts.pipelines:
-                itemproc = self.pcrawler.engine.scraper.itemproc
-                for item in items:
-                    itemproc.process_item(item, spider)
-            self.add_items(depth, items)
-            self.add_requests(depth, requests)
-
-            scraped_data = items if opts.output else []
-            if depth < opts.depth:
-                for req in requests:
-                    req.meta['_depth'] = depth + 1
-                    req.meta['_callback'] = req.callback
-                    req.callback = callback
-                scraped_data += requests
-
-            return scraped_data
+            d = self.run_callback(response, cb, cb_kwargs)
+            d.addCallback(self._get_items_and_requests, opts, depth, spider, callback)
+            d.addCallback(self.scraped_data)
+            return d
 
         # update request meta if any extra meta was passed through the --meta/-m opts.
         if opts.meta:

tests/test_command_parse.py

@@ -29,7 +29,15 @@ class ParseCommandTest(ProcessTest, SiteTest, CommandTest):
 import scrapy
 from scrapy.linkextractors import LinkExtractor
 from scrapy.spiders import CrawlSpider, Rule
+from scrapy.utils.test import get_from_asyncio_queue
 
+class AsyncDefAsyncioSpider(scrapy.Spider):
+
+    name = 'asyncdef{self.spider_name}'
+
+    async def parse(self, response):
+        status = await get_from_asyncio_queue(response.status)
+        return [scrapy.Item(), dict(foo='bar')]
 
 class MySpider(scrapy.Spider):
     name = '{self.spider_name}'
@@ -160,6 +168,13 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}}
                                            self.url('/html')])
         self.assertIn("INFO: It Works!", _textmode(stderr))
 
+    @defer.inlineCallbacks
+    def test_asyncio_parse_items(self):
+        status, out, stderr = yield self.execute(
+            ['--spider', 'asyncdef' + self.spider_name, '-c', 'parse', self.url('/html')]
+        )
+        self.assertIn("""[{}, {'foo': 'bar'}]""", _textmode(out))
+
     @defer.inlineCallbacks
     def test_parse_items(self):
         status, out, stderr = yield self.execute(