From 3faef2d08277f43659400470ec408ee29400018a Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Thu, 12 Sep 2019 20:10:58 +0500 Subject: [PATCH 1/3] Add async def support to signal handlers that already supported Deferreds. --- scrapy/utils/defer.py | 17 +++++++++++++++++ scrapy/utils/signal.py | 6 ++++-- tests/test_utils_signal.py | 26 +++++++++++++++++++++++--- 3 files changed, 44 insertions(+), 5 deletions(-) diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index bbd5ebe52..a2c24e5fb 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -140,3 +140,20 @@ 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 maybeDeferred_coro(f, *args, **kw): + """ Copy of defer.maybeDeferred that also converts coroutines to Deferreds. """ + try: + result = f(*args, **kw) + except: # noqa: E722 + return defer.fail(failure.Failure(captureVars=defer.Deferred.debug)) + + if isinstance(result, defer.Deferred): + return result + elif _isfuture(result) or inspect.isawaitable(result): + return deferred_from_coro(result) + elif isinstance(result, failure.Failure): + return defer.fail(result) + else: + return defer.succeed(result) diff --git a/scrapy/utils/signal.py b/scrapy/utils/signal.py index de00bac49..60c561da6 100644 --- a/scrapy/utils/signal.py +++ b/scrapy/utils/signal.py @@ -2,12 +2,14 @@ import logging -from twisted.internet.defer import maybeDeferred, DeferredList, Deferred +from twisted.internet.defer import DeferredList, Deferred from twisted.python.failure import Failure from pydispatch.dispatcher import Any, Anonymous, liveReceivers, \ getAllReceivers, disconnect from pydispatch.robustapply import robustApply + +from scrapy.utils.defer import maybeDeferred_coro from scrapy.utils.log import failure_to_exc_info logger = logging.getLogger(__name__) @@ -61,7 +63,7 @@ def send_catch_log_deferred(signal=Any, sender=Anonymous, *arguments, **named): spider = named.get('spider', None) dfds = [] for receiver in liveReceivers(getAllReceivers(sender, signal)): - d = maybeDeferred(robustApply, receiver, signal=signal, sender=sender, + d = maybeDeferred_coro(robustApply, receiver, signal=signal, sender=sender, *arguments, **named) d.addErrback(logerror, receiver) d.addBoth(lambda result: (receiver, result)) diff --git a/tests/test_utils_signal.py b/tests/test_utils_signal.py index 16b7c5c68..e5f6f0ed4 100644 --- a/tests/test_utils_signal.py +++ b/tests/test_utils_signal.py @@ -1,3 +1,6 @@ +import asyncio + +from pytest import mark from testfixtures import LogCapture from twisted.trial import unittest from twisted.python.failure import Failure @@ -5,6 +8,7 @@ from twisted.internet import defer, reactor from pydispatch import dispatcher from scrapy.utils.signal import send_catch_log, send_catch_log_deferred +from scrapy.utils.test import get_from_asyncio_queue class SendCatchLogTest(unittest.TestCase): @@ -54,7 +58,7 @@ class SendCatchLogDeferredTest(SendCatchLogTest): return send_catch_log_deferred(signal, *a, **kw) -class SendCatchLogDeferredTest2(SendCatchLogTest): +class SendCatchLogDeferredTest2(SendCatchLogDeferredTest): def ok_handler(self, arg, handlers_called): handlers_called.add(self.ok_handler) @@ -63,8 +67,24 @@ class SendCatchLogDeferredTest2(SendCatchLogTest): reactor.callLater(0, d.callback, "OK") return d - def _get_result(self, signal, *a, **kw): - return send_catch_log_deferred(signal, *a, **kw) + +class SendCatchLogDeferredAsyncDefTest(SendCatchLogDeferredTest): + + async def ok_handler(self, arg, handlers_called): + handlers_called.add(self.ok_handler) + assert arg == 'test' + await defer.succeed(42) + return "OK" + + +@mark.only_asyncio() +class SendCatchLogDeferredAsyncioTest(SendCatchLogDeferredTest): + + async def ok_handler(self, arg, handlers_called): + handlers_called.add(self.ok_handler) + assert arg == 'test' + await asyncio.sleep(0.2) + return await get_from_asyncio_queue("OK") class SendCatchLogTest2(unittest.TestCase): From 489ffcda5143a2ef28d4cbcf5418babd963f2b0f Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Thu, 6 Feb 2020 22:39:00 +0500 Subject: [PATCH 2/3] Add a test for an async item_scraped handler. --- tests/test_signals.py | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 tests/test_signals.py diff --git a/tests/test_signals.py b/tests/test_signals.py new file mode 100644 index 000000000..001e798e5 --- /dev/null +++ b/tests/test_signals.py @@ -0,0 +1,39 @@ +from twisted.internet import defer +from twisted.trial import unittest + +from scrapy import signals, Request, Spider +from scrapy.utils.test import get_crawler + +from tests.mockserver import MockServer + + +class ItemSpider(Spider): + name = 'itemspider' + + def start_requests(self): + for _ in range(10): + yield Request(self.mockserver.url('/status?n=200'), + dont_filter=True) + + def parse(self, response): + return {'field': 42} + + +class AsyncSignalTestCase(unittest.TestCase): + def setUp(self): + self.mockserver = MockServer() + self.mockserver.__enter__() + self.items = [] + + def tearDown(self): + self.mockserver.__exit__(None, None, None) + + async def _on_item_scraped(self, item): + self.items.append(item) + + @defer.inlineCallbacks + def test_simple_pipeline(self): + crawler = get_crawler(ItemSpider) + crawler.signals.connect(self._on_item_scraped, signals.item_scraped) + yield crawler.crawl(mockserver=self.mockserver) + self.assertEqual(len(self.items), 10) From 1f0f52cbf7bdc9f11f7b83c482ad52ad7ad32ba0 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 11 Feb 2020 01:05:45 +0500 Subject: [PATCH 3/3] Improve async signal tests. --- tests/test_signals.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/test_signals.py b/tests/test_signals.py index 001e798e5..d6ae526be 100644 --- a/tests/test_signals.py +++ b/tests/test_signals.py @@ -1,8 +1,9 @@ +from pytest import mark from twisted.internet import defer from twisted.trial import unittest from scrapy import signals, Request, Spider -from scrapy.utils.test import get_crawler +from scrapy.utils.test import get_crawler, get_from_asyncio_queue from tests.mockserver import MockServer @@ -11,12 +12,12 @@ class ItemSpider(Spider): name = 'itemspider' def start_requests(self): - for _ in range(10): - yield Request(self.mockserver.url('/status?n=200'), - dont_filter=True) + for index in range(10): + yield Request(self.mockserver.url('/status?n=200&id=%d' % index), + meta={'index': index}) def parse(self, response): - return {'field': 42} + return {'index': response.meta['index']} class AsyncSignalTestCase(unittest.TestCase): @@ -29,11 +30,15 @@ class AsyncSignalTestCase(unittest.TestCase): self.mockserver.__exit__(None, None, None) async def _on_item_scraped(self, item): + item = await get_from_asyncio_queue(item) self.items.append(item) + @mark.only_asyncio() @defer.inlineCallbacks def test_simple_pipeline(self): crawler = get_crawler(ItemSpider) crawler.signals.connect(self._on_item_scraped, signals.item_scraped) yield crawler.crawl(mockserver=self.mockserver) self.assertEqual(len(self.items), 10) + for index in range(10): + self.assertIn({'index': index}, self.items)