From 81175669746737f1ab0bd0236a70f787ed1855a4 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Mon, 16 Dec 2019 22:12:27 +0500 Subject: [PATCH 1/6] Add utils.defer.deferred_f_from_coro_f. --- scrapy/utils/defer.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index bbd5ebe52..62b43a96c 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -2,6 +2,7 @@ Helper functions for dealing with Twisted deferreds """ import asyncio +from functools import wraps import inspect from twisted.internet import defer, task @@ -140,3 +141,15 @@ def deferred_from_coro(o): # wrapping the coroutine into a Future and then into a Deferred, this requires AsyncioSelectorReactor return defer.Deferred.fromFuture(asyncio.ensure_future(o)) return o + + +def deferred_f_from_coro_f(coro_f): + """ Converts a coroutine function into a function that returns a Deferred. + + The coroutine function will be called at the time when the wrapper is called. Wrapper args will be passed to it. + This is useful for callback chains, as callback functions are called with the previous callback result. + """ + @wraps(coro_f) + def f(*coro_args, **coro_kwargs): + return deferred_from_coro(coro_f(*coro_args, **coro_kwargs)) + return f From 1f9cef787d3ca0c12099f1b1b4c52efc510e381d Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 10 Sep 2019 14:26:21 +0500 Subject: [PATCH 2/6] Add async def support to pipelines. --- scrapy/pipelines/__init__.py | 4 +++- tests/test_pipelines.py | 13 +++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/scrapy/pipelines/__init__.py b/scrapy/pipelines/__init__.py index aa1bfb77f..1a45e00a2 100644 --- a/scrapy/pipelines/__init__.py +++ b/scrapy/pipelines/__init__.py @@ -6,6 +6,8 @@ See documentation in docs/item-pipeline.rst from scrapy.middleware import MiddlewareManager from scrapy.utils.conf import build_component_list +from scrapy.utils.defer import deferred_f_from_coro_f + class ItemPipelineManager(MiddlewareManager): @@ -19,7 +21,7 @@ class ItemPipelineManager(MiddlewareManager): def _add_middleware(self, pipe): super(ItemPipelineManager, self)._add_middleware(pipe) if hasattr(pipe, 'process_item'): - self.methods['process_item'].append(pipe.process_item) + self.methods['process_item'].append(deferred_f_from_coro_f(pipe.process_item)) def process_item(self, item, spider): return self._process_chain('process_item', item, spider) diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py index bc53f5427..cfe4471d7 100644 --- a/tests/test_pipelines.py +++ b/tests/test_pipelines.py @@ -26,6 +26,13 @@ class DeferredPipeline: return d +class AsyncDefPipeline: + async def process_item(self, item, spider): + await defer.succeed(42) + item['pipeline_passed'] = True + return item + + class ItemSpider(Spider): name = 'itemspider' @@ -69,3 +76,9 @@ class PipelineTestCase(unittest.TestCase): crawler = self._create_crawler(DeferredPipeline) yield crawler.crawl(mockserver=self.mockserver) self.assertEqual(len(self.items), 1) + + @defer.inlineCallbacks + def test_asyncdef_pipeline(self): + crawler = self._create_crawler(AsyncDefPipeline) + yield crawler.crawl(mockserver=self.mockserver) + self.assertEqual(len(self.items), 1) From bfdd552a32b3a67dfa2895b483c187d87b63b50e Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 10 Sep 2019 14:57:07 +0500 Subject: [PATCH 3/6] Add a test for pipelines using asyncio. --- tests/test_pipelines.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py index cfe4471d7..aba6d85b7 100644 --- a/tests/test_pipelines.py +++ b/tests/test_pipelines.py @@ -1,3 +1,5 @@ +import asyncio + from twisted.internet import defer from twisted.internet.defer import Deferred from twisted.trial import unittest @@ -33,6 +35,13 @@ class AsyncDefPipeline: return item +class AsyncDefAsyncioPipeline: + async def process_item(self, item, spider): + await asyncio.sleep(0.2) + item['pipeline_passed'] = True + return item + + class ItemSpider(Spider): name = 'itemspider' @@ -82,3 +91,9 @@ class PipelineTestCase(unittest.TestCase): crawler = self._create_crawler(AsyncDefPipeline) yield crawler.crawl(mockserver=self.mockserver) self.assertEqual(len(self.items), 1) + + @defer.inlineCallbacks + def test_asyncdef_asyncio_pipeline(self): + crawler = self._create_crawler(AsyncDefAsyncioPipeline) + yield crawler.crawl(mockserver=self.mockserver) + self.assertEqual(len(self.items), 1) From bdef948aaebced3d28ea7b81525ea9edf7cc650c Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Mon, 16 Dec 2019 23:15:43 +0500 Subject: [PATCH 4/6] Mark the asyncio pipelines test as only_asyncio. --- tests/test_pipelines.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py index aba6d85b7..6f33282f2 100644 --- a/tests/test_pipelines.py +++ b/tests/test_pipelines.py @@ -1,5 +1,6 @@ import asyncio +from pytest import mark from twisted.internet import defer from twisted.internet.defer import Deferred from twisted.trial import unittest @@ -92,6 +93,7 @@ class PipelineTestCase(unittest.TestCase): yield crawler.crawl(mockserver=self.mockserver) self.assertEqual(len(self.items), 1) + @mark.only_asyncio() @defer.inlineCallbacks def test_asyncdef_asyncio_pipeline(self): crawler = self._create_crawler(AsyncDefAsyncioPipeline) From 9d8c54c0f2a98dc627efe8a2b58cfd311f2105cb Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Mon, 16 Dec 2019 22:43:55 +0500 Subject: [PATCH 5/6] Fix/ignore flake8 problems. --- pytest.ini | 1 + scrapy/pipelines/__init__.py | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/pytest.ini b/pytest.ini index c3f3292bb..1030d5530 100644 --- a/pytest.ini +++ b/pytest.ini @@ -95,6 +95,7 @@ flake8-ignore = scrapy/loader/__init__.py E501 E128 scrapy/loader/processors.py E501 # scrapy/pipelines + scrapy/pipelines/__init__.py E501 scrapy/pipelines/files.py E116 E501 E266 scrapy/pipelines/images.py E265 E501 scrapy/pipelines/media.py E125 E501 E266 diff --git a/scrapy/pipelines/__init__.py b/scrapy/pipelines/__init__.py index 1a45e00a2..b5725a8ee 100644 --- a/scrapy/pipelines/__init__.py +++ b/scrapy/pipelines/__init__.py @@ -9,7 +9,6 @@ from scrapy.utils.conf import build_component_list from scrapy.utils.defer import deferred_f_from_coro_f - class ItemPipelineManager(MiddlewareManager): component_name = 'item pipeline' From 7d859848800ef05760e4f57dc206a4d756460a9d Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Thu, 9 Jan 2020 14:48:07 +0500 Subject: [PATCH 6/6] Use get_from_asyncio_queue in the pipeline test. --- tests/test_pipelines.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py index 6f33282f2..c72f1a338 100644 --- a/tests/test_pipelines.py +++ b/tests/test_pipelines.py @@ -6,7 +6,7 @@ from twisted.internet.defer import Deferred from twisted.trial import unittest from scrapy import Spider, signals, Request -from scrapy.utils.test import get_crawler +from scrapy.utils.test import get_crawler, get_from_asyncio_queue from tests.mockserver import MockServer @@ -39,7 +39,7 @@ class AsyncDefPipeline: class AsyncDefAsyncioPipeline: async def process_item(self, item, spider): await asyncio.sleep(0.2) - item['pipeline_passed'] = True + item['pipeline_passed'] = await get_from_asyncio_queue(True) return item