Add Deferred-to-Future helpers (#5288)

conftest.py

@@ -75,6 +75,12 @@ def only_asyncio(request, reactor_pytest):
         pytest.skip('This test is only run with --reactor=asyncio')
 
 
+@pytest.fixture(autouse=True)
+def only_not_asyncio(request, reactor_pytest):
+    if request.node.get_closest_marker('only_not_asyncio') and reactor_pytest == 'asyncio':
+        pytest.skip('This test is only run without --reactor=asyncio')
+
+
 def pytest_configure(config):
     if config.getoption("--reactor") == "asyncio":
         install_reactor("twisted.internet.asyncioreactor.AsyncioSelectorReactor")

docs/topics/asyncio.rst

@@ -39,5 +39,22 @@ You can also use custom asyncio event loops with the asyncio reactor. Set the
 :setting:`ASYNCIO_EVENT_LOOP` setting to the import path of the desired event loop class to
 use it instead of the default asyncio event loop.
 
+.. _asyncio-await-dfd:
 
+Awaiting on Deferreds
+=====================
 
+When the asyncio reactor isn't installed, you can await on Deferreds in the
+coroutines directly. When it is installed, this is not possible anymore, due to
+specifics of the Scrapy coroutine integration (the coroutines are wrapped into
+:class:`asyncio.Future` objects, not into
+:class:`~twisted.internet.defer.Deferred` directly), and you need to wrap them into
+Futures. Scrapy provides two helpers for this:
+
+.. autofunction:: scrapy.utils.defer.deferred_to_future
+.. autofunction:: scrapy.utils.defer.maybe_deferred_to_future
+.. tip:: If you need to use these functions in code that aims to be compatible
+         with lower versions of Scrapy that do not provide these functions,
+         down to Scrapy 2.0 (earlier versions do not support
+         :mod:`asyncio`), you can copy the implementation of these functions
+         into your own code.

docs/topics/coroutines.rst

@@ -75,23 +75,28 @@ coroutines, functions that return Deferreds and functions that return
 :term:`awaitable objects <awaitable>` such as :class:`~asyncio.Future`.
 This means you can use many useful Python libraries providing such code::
 
-    class MySpider(Spider):
+    class MySpiderDeferred(Spider):
         # ...
-        async def parse_with_deferred(self, response):
+        async def parse(self, response):
             additional_response = await treq.get('https://additional.url')
             additional_data = await treq.content(additional_response)
             # ... use response and additional_data to yield items and requests
 
-        async def parse_with_asyncio(self, response):
+    class MySpiderAsyncio(Spider):
+        # ...
+        async def parse(self, response):
             async with aiohttp.ClientSession() as session:
                 async with session.get('https://additional.url') as additional_response:
-                    additional_data = await r.text()
+                    additional_data = await additional_response.text()
             # ... use response and additional_data to yield items and requests
 
 .. note:: Many libraries that use coroutines, such as `aio-libs`_, require the
           :mod:`asyncio` loop and to use them you need to
           :doc:`enable asyncio support in Scrapy<asyncio>`.
 
+.. note:: If you want to ``await`` on Deferreds while using the asyncio reactor,
+          you need to :ref:`wrap them<asyncio-await-dfd>`.
+
 Common use cases for asynchronous code include:
 
 * requesting data from websites, databases and other services (in callbacks,

docs/topics/item-pipeline.rst

@@ -190,6 +190,8 @@ item.
 
     import scrapy
     from itemadapter import ItemAdapter
+    from scrapy.utils.defer import maybe_deferred_to_future
+
 
     class ScreenshotPipeline:
         """Pipeline that uses Splash to render screenshot of
@@ -202,7 +204,7 @@ item.
             encoded_item_url = quote(adapter["url"])
             screenshot_url = self.SPLASH_URL.format(encoded_item_url)
             request = scrapy.Request(screenshot_url)
-            response = await spider.crawler.engine.download(request, spider)
+            response = await maybe_deferred_to_future(spider.crawler.engine.download(request, spider))
 
             if response.status != 200:
                 # Error happened, return item.

pytest.ini

@@ -20,5 +20,6 @@ addopts =
     --ignore=docs/utils
 markers =
     only_asyncio: marks tests as only enabled when --reactor=asyncio is passed
+    only_not_asyncio: marks tests as only enabled when --reactor=asyncio is not passed
 filterwarnings=
     ignore::DeprecationWarning:twisted.web.test.test_webclient

scrapy/utils/defer.py

@@ -3,9 +3,16 @@ Helper functions for dealing with Twisted deferreds
 """
 import asyncio
 import inspect
-from collections.abc import Coroutine
+from asyncio import Future
 from functools import wraps
-from typing import Any, Callable, Generator, Iterable
+from typing import (
+    Any,
+    Callable,
+    Coroutine,
+    Generator,
+    Iterable,
+    Union
+)
 
 from twisted.internet import defer
 from twisted.internet.defer import Deferred, DeferredList, ensureDeferred
@@ -171,3 +178,55 @@ def maybeDeferred_coro(f: Callable, *args, **kw) -> Deferred:
         return defer.fail(result)
     else:
         return defer.succeed(result)
+
+
+def deferred_to_future(d: Deferred) -> Future:
+    """
+    .. versionadded:: VERSION
+
+    Return an :class:`asyncio.Future` object that wraps *d*.
+
+    When :ref:`using the asyncio reactor <install-asyncio>`, you cannot await
+    on :class:`~twisted.internet.defer.Deferred` objects from :ref:`Scrapy
+    callables defined as coroutines <coroutine-support>`, you can only await on
+    ``Future`` objects. Wrapping ``Deferred`` objects into ``Future`` objects
+    allows you to wait on them::
+
+        class MySpider(Spider):
+            ...
+            async def parse(self, response):
+                d = treq.get('https://example.com/additional')
+                additional_response = await deferred_to_future(d)
+    """
+    return d.asFuture(asyncio.get_event_loop())
+
+
+def maybe_deferred_to_future(d: Deferred) -> Union[Deferred, Future]:
+    """
+    .. versionadded:: VERSION
+
+    Return *d* as an object that can be awaited from a :ref:`Scrapy callable
+    defined as a coroutine <coroutine-support>`.
+
+    What you can await in Scrapy callables defined as coroutines depends on the
+    value of :setting:`TWISTED_REACTOR`:
+
+    -   When not using the asyncio reactor, you can only await on
+        :class:`~twisted.internet.defer.Deferred` objects.
+
+    -   When :ref:`using the asyncio reactor <install-asyncio>`, you can only
+        await on :class:`asyncio.Future` objects.
+
+    If you want to write code that uses ``Deferred`` objects but works with any
+    reactor, use this function on all ``Deferred`` objects::
+
+        class MySpider(Spider):
+            ...
+            async def parse(self, response):
+                d = treq.get('https://example.com/additional')
+                extra_response = await maybe_deferred_to_future(d)
+    """
+    if not is_asyncio_reactor_installed():
+        return d
+    else:
+        return deferred_to_future(d)

tests/test_pipelines.py

@@ -6,6 +6,7 @@ from twisted.internet.defer import Deferred
 from twisted.trial import unittest
 
 from scrapy import Spider, signals, Request
+from scrapy.utils.defer import maybe_deferred_to_future, deferred_to_future
 from scrapy.utils.test import get_crawler, get_from_asyncio_queue
 
 from tests.mockserver import MockServer
@@ -31,18 +32,38 @@ class DeferredPipeline:
 
 class AsyncDefPipeline:
     async def process_item(self, item, spider):
-        await defer.succeed(42)
+        d = Deferred()
+        from twisted.internet import reactor
+        reactor.callLater(0, d.callback, None)
+        await maybe_deferred_to_future(d)
         item['pipeline_passed'] = True
         return item
 
 
 class AsyncDefAsyncioPipeline:
     async def process_item(self, item, spider):
+        d = Deferred()
+        from twisted.internet import reactor
+        reactor.callLater(0, d.callback, None)
+        await deferred_to_future(d)
         await asyncio.sleep(0.2)
         item['pipeline_passed'] = await get_from_asyncio_queue(True)
         return item
 
 
+class AsyncDefNotAsyncioPipeline:
+    async def process_item(self, item, spider):
+        d1 = Deferred()
+        from twisted.internet import reactor
+        reactor.callLater(0, d1.callback, None)
+        await d1
+        d2 = Deferred()
+        reactor.callLater(0, d2.callback, None)
+        await maybe_deferred_to_future(d2)
+        item['pipeline_passed'] = True
+        return item
+
+
 class ItemSpider(Spider):
     name = 'itemspider'
 
@@ -99,3 +120,10 @@ class PipelineTestCase(unittest.TestCase):
         crawler = self._create_crawler(AsyncDefAsyncioPipeline)
         yield crawler.crawl(mockserver=self.mockserver)
         self.assertEqual(len(self.items), 1)
+
+    @mark.only_not_asyncio()
+    @defer.inlineCallbacks
+    def test_asyncdef_not_asyncio_pipeline(self):
+        crawler = self._create_crawler(AsyncDefNotAsyncioPipeline)
+        yield crawler.crawl(mockserver=self.mockserver)
+        self.assertEqual(len(self.items), 1)