Implement spider middleware iterable upgrade/downgrade.

docs/topics/coroutines.rst

@@ -35,11 +35,10 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
 -   :ref:`Signal handlers that support deferreds <signal-deferred>`.
 
 -   The :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output`
-    method of :ref:`spider middlewares <custom-spider-middleware>`.
+    method of :ref:`spider middlewares <custom-spider-middleware>`. See
+    :ref:`async-spider-middlewares`.
 
     .. versionadded:: VERSION
-    .. note:: This method needs to be an async generator, not just a coroutine that
-              returns an iterable.
 
 Usage
 =====
@@ -106,8 +105,8 @@ Common use cases for asynchronous code include:
 * storing data in databases (in pipelines and middlewares);
 * delaying the spider initialization until some external event (in the
   :signal:`spider_opened` handler);
-* calling asynchronous Scrapy methods like ``ExecutionEngine.download`` (see
-  :ref:`the screenshot pipeline example<ScreenshotPipeline>`).
+* calling asynchronous Scrapy methods like :meth:`ExecutionEngine.download`
+  (see :ref:`the screenshot pipeline example<ScreenshotPipeline>`).
 
 .. _aio-libs: https://github.com/aio-libs
 
@@ -119,59 +118,72 @@ Asynchronous spider middlewares
 .. versionadded:: VERSION
 .. note:: This currently applies to
           :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output`.
+          In the future it will also apply to
+          :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_start_requests`.
 
 Middleware methods discussed here can take and return async iterables. They can
 return the same type of iterable or they can take a normal one and return an
 async one. If such method needs to return an async iterable it must be an async
 generator, not just a coroutine that returns an iterable.
 
-.. autofunction:: scrapy.utils.asyncgen.as_async_generator
+As the result of a middleware method is passed to the same method of the next
+middleware, it needs to be adapted if the second method expects a different
+type. Scrapy will do this transparently:
 
-In the simplest form that supports both sync and async input it can be written
-like this::
+* A normal iterable is wrapped into an async one which shouldn't cause any side
+  effects.
+* An async iterable is downgraded to a normal one by waiting until all results
+  are available and wrapping them in a normal iterable. This is problematic
+  because it pauses the normal middleware processing for this iterable and
+  because all results can be skipped if exceptions are raised during
+  processing. This case emits a warning and will be deprecated and then removed
+  in a later Scrapy version.
+* Async iterables returned from
+  :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_exception`
+  won't be downgraded, an exception will be raised if that is needed.
 
-    from scrapy.utils.asyncgen import as_async_generator
+As downgrading is undesirable, here is the proposed way to avoid it. If all
+middlewares, including 3rd-party ones, support async iterables as input, no
+downgrading will happen. But removing normal iterable support (making the
+method a coroutine) from a middleware published as a separate project or used
+internally in projects for older Scrapy versions breaks backwards
+compatibility. So, as an interim measure (it will be deprecated and then
+removed in a later Scrapy version), a middleware can provide both sync and
+async methods in the following form::
 
-    class ProcessSpiderOutputAsyncGenMiddleware:
-        async def process_spider_output(self, response, result, spider):
-            async for r in as_async_generator(result):
+    class UniversalSpiderMiddleware:
+        def process_spider_output(self, response, result, spider):
+            for r in result:
                 # ... do something with r
                 yield r
 
-If the middleware input (the callback result for ``process_spider_output``) is
-an async iterable, all middlewares that process it must support it. The
-built-in ones do, but the ones in your project and 3rd-party ones will need to
-be updated to support it, as the code that expects a normal iterable will break
-on an async one. If these middlewares receive an async iterable, they must
-return one as well. On the other hand, if they receive a normal iterable, they
-shouldn't break and ideally should return a normal iterable too. There can be
-several possible implementations of this.
+        async def process_spider_output_async(self, response, result, spider):
+            async for r in result:
+                # ... do something with r
+                yield r
 
-The simplest one, always converting normal iterables to async ones, is provided
-above. Because a result of a middleware method is passed to the same method of
-the next middleware, it's only possible to mix middlewares with synchronous and
-asynchronous implementations of the same method if all synchronous ones are
-called first (which isn't always possible).
+In this case normal and async iterables will be passed to the respective
+methods without any wrapping or downgrading, and in older versions of Scrapy
+the coroutine method will just be ignored. When the backwards compatibility is
+no longer needed the non-coroutine method can be dropped and the coroutine one
+renamed to the normal name. It may be possible to extract common code from both
+methods to reduce code duplication, as in the simplest case the only difference
+between them will be ``for`` vs ``async for``.
 
-Another option is to make separate methods for normal and async iterables and
-choose one at run time::
+So, to recap:
 
-    from inspect import isasyncgen
-
-    class ProcessSpiderOutputAsyncGenMiddleware:
-        def _normal_process_spider_output(self, response, result, spider):
-            # ... do something with normal result
-
-        async def _async_process_spider_output(self, response, result, spider):
-            # ... do the same with async result
-
-        def process_spider_output(self, response, result, spider):
-            if isasyncgen(result):
-                return self._async_process_spider_output(self, response, result, spider)
-            else:
-                return self._normal_process_spider_output(self, response, result, spider)
-
-If you are writing a middleware that you intend to publish or to use in many
-projects, this is likely the best way to implement it. It may be possible to
-extract common code from both methods to reduce code duplication, as in the
-simplest case the only difference between them will be ``for`` vs ``async for``.
+* If you don't intend to use async callbacks or middlewares containing async
+  code in your project, nothing should change for you yet. At some point in the
+  future some of the 3rd-party middlewares you use may drop backwards
+  compatibility, which shouldn't lead to immediate problems but may be a sign
+  to start converting your code to ``async def`` too.
+* If you maintain a middleware that can be used with projects you can't control
+  (e.g. one you published for other people to use, or one that needs to support
+  some old project that can't be modernized), we recommend adding a
+  ``process_spider_output_async`` method so that the amount of unnecessary
+  iterable conversions is reduced but no compatibility is broken.
+* If you use async callbacks, try to make sure all middlewares support them.
+  Note that you can modernize 3rd-party middlewares by subclassing them.
+* If you want to write and publish a middleware that requires async code, you
+  should write in the docs that the minimum support Scrapy version is VERSION
+  (maybe even check this at the run time, using :attr:`scrapy.__version__`).

docs/topics/spider-middleware.rst

@@ -123,6 +123,15 @@ object gives you access, for example, to the :ref:`settings <topics-settings>`.
         :param spider: the spider whose result is being processed
         :type spider: :class:`~scrapy.Spider` object
 
+    .. method:: process_spider_output_async(response, result, spider)
+
+        .. versionadded:: VERSION
+
+        If exists, this methid will be called instead of
+        :meth:`process_spider_output` when ``result`` is an async iterable.
+        If this method exists, it must be a coroutine while
+        :meth:`process_spider_output` must not be a coroutine.
+
     .. method:: process_spider_exception(response, exception, spider)
 
         This method is called when a spider or :meth:`process_spider_output`

scrapy/core/scraper.py

@@ -192,7 +192,8 @@ class Scraper:
             spider=spider
         )
 
-    def handle_spider_output(self, result: Iterable, request: Request, response: Response, spider: Spider) -> Deferred:
+    def handle_spider_output(self, result: Union[Iterable, AsyncIterable], request: Request,
+                             response: Response, spider: Spider) -> Deferred:
         if not result:
             return defer_succeed(None)
         it: Union[Generator, AsyncGenerator]

scrapy/core/spidermw.py

@@ -3,22 +3,27 @@ Spider Middleware manager
 
 See documentation in docs/topics/spider-middleware.rst
 """
+import logging
+from inspect import isasyncgenfunction
 from itertools import islice
-from typing import Any, AsyncGenerator, AsyncIterable, Callable, Generator, Iterable, Union, cast
+from typing import Any, AsyncGenerator, AsyncIterable, Callable, Generator, Iterable, Tuple, Union, cast
 
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
 from twisted.python.failure import Failure
 
 from scrapy import Request, Spider
 from scrapy.exceptions import _InvalidOutput
 from scrapy.http import Response
 from scrapy.middleware import MiddlewareManager
-from scrapy.utils.asyncgen import _process_iterable_universal
+from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.conf import build_component_list
-from scrapy.utils.defer import mustbe_deferred
+from scrapy.utils.defer import mustbe_deferred, deferred_from_coro, deferred_f_from_coro_f, maybe_deferred_to_future
 from scrapy.utils.python import MutableAsyncChain, MutableChain
 
 
+logger = logging.getLogger(__name__)
+
+
 ScrapeFunc = Callable[[Union[Response, Failure], Request, Spider], Any]
 
 
@@ -30,6 +35,10 @@ class SpiderMiddlewareManager(MiddlewareManager):
 
     component_name = 'spider middleware'
 
+    def __init__(self, *middlewares):
+        super().__init__(*middlewares)
+        self.downgrade_warning_done = False
+
     @classmethod
     def _get_mwlist_from_settings(cls, settings):
         return build_component_list(settings.getwithbase('SPIDER_MIDDLEWARES'))
@@ -40,7 +49,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
             self.methods['process_spider_input'].append(mw.process_spider_input)
         if hasattr(mw, 'process_start_requests'):
             self.methods['process_start_requests'].appendleft(mw.process_start_requests)
-        process_spider_output = getattr(mw, 'process_spider_output', None)
+        process_spider_output = self._get_async_method_pair(mw, 'process_spider_output')
         self.methods['process_spider_output'].appendleft(process_spider_output)
         process_spider_exception = getattr(mw, 'process_spider_exception', None)
         self.methods['process_spider_exception'].appendleft(process_spider_exception)
@@ -64,8 +73,19 @@ class SpiderMiddlewareManager(MiddlewareManager):
     def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Union[Iterable, AsyncIterable],
                            exception_processor_index: int, recover_to: Union[MutableChain, MutableAsyncChain]
                            ) -> Union[Generator, AsyncGenerator]:
-        @_process_iterable_universal
-        async def _evaluate_async_iterable(iterable):
+
+        def process_sync(iterable: Iterable):
+            try:
+                for r in iterable:
+                    yield r
+            except Exception as ex:
+                exception_result = self._process_spider_exception(response, spider, Failure(ex),
+                                                                  exception_processor_index)
+                if isinstance(exception_result, Failure):
+                    raise
+                recover_to.extend(exception_result)
+
+        async def process_async(iterable: AsyncIterable):
             try:
                 async for r in iterable:
                     yield r
@@ -75,7 +95,10 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 if isinstance(exception_result, Failure):
                     raise
                 recover_to.extend(exception_result)
-        return _evaluate_async_iterable(iterable)
+
+        if isinstance(iterable, AsyncIterable):
+            return process_async(iterable)
+        return process_sync(iterable)
 
     def _process_spider_exception(self, response: Response, spider: Spider, _failure: Failure,
                                   start_index: int = 0) -> Union[Failure, MutableChain]:
@@ -87,11 +110,22 @@ class SpiderMiddlewareManager(MiddlewareManager):
         for method_index, method in enumerate(method_list, start=start_index):
             if method is None:
                 continue
+            method = cast(Callable, method)
             result = method(response=response, exception=exception, spider=spider)
             if _isiterable(result):
                 # stop exception handling by handing control over to the
                 # process_spider_output chain if an iterable has been returned
-                return self._process_spider_output(response, spider, result, method_index + 1)
+                dfd: Deferred = self._process_spider_output(response, spider, result, method_index + 1)
+                # _process_spider_output() returns a Deferred only because of downgrading so this can be
+                # simplified when downgrading is removed.
+                if dfd.called:
+                    # the result is available immediately if _process_spider_output didn't do downgrading
+                    return dfd.result
+                else:
+                    # we forbid waiting here because otherwise we would need to return a deferred from
+                    # _process_spider_exception too, which complicates the architecture
+                    msg = f"Async iterable returned from {method.__qualname__} cannot be downgraded"
+                    raise _InvalidOutput(msg)
             elif result is None:
                 continue
             else:
@@ -100,9 +134,13 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 raise _InvalidOutput(msg)
         return _failure
 
+    # This method cannot be made async def, as _process_spider_exception relies on the Deferred result
+    # being available immediately which doesn't work when it's a wrapped coroutine.
+    # It also needs @inlineCallbacks only because of downgrading so it can be removed when downgrading is removed.
+    @inlineCallbacks
     def _process_spider_output(self, response: Response, spider: Spider,
                                result: Union[Iterable, AsyncIterable], start_index: int = 0
-                               ) -> Union[MutableChain, MutableAsyncChain]:
+                               ) -> Deferred:
         # items in this iterable do not need to go through the process_spider_output
         # chain, they went through it already from the process_spider_exception method
         recovered: Union[MutableChain, MutableAsyncChain]
@@ -112,11 +150,43 @@ class SpiderMiddlewareManager(MiddlewareManager):
         else:
             recovered = MutableChain()
 
+        # There are three cases for the middleware: def foo, async def foo, def foo + async def foo_async.
+        # 1. def foo. Sync iterables are passed as is, async ones are downgraded.
+        # 2. async def foo. Sync iterables are upgraded, async ones are passed as is.
+        # 3. def foo + async def foo_async. Iterables are passed to the respective method.
+        # Storing methods and method tuples in the same list is weird but we should be able to roll this back
+        # when we drop this compatibility feature.
+
         method_list = islice(self.methods['process_spider_output'], start_index, None)
-        for method_index, method in enumerate(method_list, start=start_index):
-            if method is None:
+        for method_index, method_pair in enumerate(method_list, start=start_index):
+            if method_pair is None:
                 continue
+            need_upgrade = need_downgrade = False
+            if isinstance(method_pair, tuple):
+                # This tuple handling is only needed until _async compatibility methods are removed.
+                method_sync, method_async = method_pair
+                method = method_async if last_result_is_async else method_sync
+            else:
+                method = method_pair
+                if not last_result_is_async and isasyncgenfunction(method):
+                    need_upgrade = True
+                elif last_result_is_async and not isasyncgenfunction(method):
+                    need_downgrade = True
             try:
+                if need_upgrade:
+                    # Iterable -> AsyncIterable
+                    result = as_async_generator(result)
+                elif need_downgrade:
+                    if not self.downgrade_warning_done:
+                        logger.warning(f"Async iterable passed to {method.__qualname__} "
+                                       f"was downgraded to a non-async one")
+                        self.downgrade_warning_done = True
+                    assert isinstance(result, AsyncIterable)
+                    # AsyncIterable -> Iterable
+                    result = yield deferred_from_coro(collect_asyncgen(result))
+                    if isinstance(recovered, AsyncIterable):
+                        recovered_collected = yield deferred_from_coro(collect_asyncgen(recovered))
+                        recovered = MutableChain(recovered_collected)
                 # might fail directly if the output value is not a generator
                 result = method(response=response, result=result, spider=spider)
             except Exception as ex:
@@ -130,8 +200,6 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 msg = (f"Middleware {method.__qualname__} must return an "
                        f"iterable, got {type(result)}")
                 raise _InvalidOutput(msg)
-            if last_result_is_async and isinstance(result, Iterable):
-                raise TypeError(f"Synchronous {method.__qualname__} called with an async iterable")
             last_result_is_async = isinstance(result, AsyncIterable)
 
         if last_result_is_async:
@@ -139,31 +207,58 @@ class SpiderMiddlewareManager(MiddlewareManager):
         else:
             return MutableChain(result, recovered)  # type: ignore[arg-type]
 
-    def _process_callback_output(self, response: Response, spider: Spider, result: Union[Iterable, AsyncIterable]
-                                 ) -> Union[MutableChain, MutableAsyncChain]:
+    async def _process_callback_output(self, response: Response, spider: Spider, result: Union[Iterable, AsyncIterable]
+                                       ) -> Union[MutableChain, MutableAsyncChain]:
         recovered: Union[MutableChain, MutableAsyncChain]
         if isinstance(result, AsyncIterable):
             recovered = MutableAsyncChain()
         else:
             recovered = MutableChain()
         result = self._evaluate_iterable(response, spider, result, 0, recovered)
-        result = self._process_spider_output(response, spider, result)
+        result = await maybe_deferred_to_future(self._process_spider_output(response, spider, result))
         if isinstance(result, AsyncIterable):
             return MutableAsyncChain(result, recovered)
         else:
+            if isinstance(recovered, AsyncIterable):
+                recovered_collected = await collect_asyncgen(recovered)
+                recovered = MutableChain(recovered_collected)
             return MutableChain(result, recovered)  # type: ignore[arg-type]
 
     def scrape_response(self, scrape_func: ScrapeFunc, response: Response, request: Request,
                         spider: Spider) -> Deferred:
-        def process_callback_output(result: Union[Iterable, AsyncIterable]) -> Union[MutableChain, MutableAsyncChain]:
-            return self._process_callback_output(response, spider, result)
+        async def process_callback_output(result: Union[Iterable, AsyncIterable]
+                                          ) -> Union[MutableChain, MutableAsyncChain]:
+            return await self._process_callback_output(response, spider, result)
 
         def process_spider_exception(_failure: Failure) -> Union[Failure, MutableChain]:
             return self._process_spider_exception(response, spider, _failure)
 
         dfd = mustbe_deferred(self._process_spider_input, scrape_func, response, request, spider)
-        dfd.addCallbacks(callback=process_callback_output, errback=process_spider_exception)
+        dfd.addCallbacks(callback=deferred_f_from_coro_f(process_callback_output), errback=process_spider_exception)
         return dfd
 
     def process_start_requests(self, start_requests, spider: Spider) -> Deferred:
         return self._process_chain('process_start_requests', start_requests, spider)
+
+    # This method is only needed until _async compatibility methods are removed.
+    @staticmethod
+    def _get_async_method_pair(mw: Any, methodname: str) -> Union[None, Callable, Tuple[Callable, Callable]]:
+        normal_method = getattr(mw, methodname, None)
+        methodname_async = methodname + "_async"
+        async_method = getattr(mw, methodname_async, None)
+        if not async_method:
+            return normal_method
+        if not normal_method:
+            logger.error(f"Middleware {mw.__qualname__} has {methodname_async} "
+                         f"without {methodname}, skipping this method.")
+            return None
+        if not isasyncgenfunction(async_method):
+            logger.error(f"{async_method.__qualname__} is not "
+                         f"an async generator function, skipping this method.")
+            return normal_method
+        if isasyncgenfunction(normal_method):
+            logger.error(f"{normal_method.__qualname__} is an async "
+                         f"generator function while {methodname_async} exists, "
+                         f"skipping both methods.")
+            return None
+        return normal_method, async_method

scrapy/middleware.py

@@ -1,7 +1,7 @@
 import logging
 import pprint
 from collections import defaultdict, deque
-from typing import Callable, Deque, Dict, Optional, cast, Iterable
+from typing import Callable, Deque, Dict, Iterable, Tuple, Union, cast
 
 from twisted.internet.defer import Deferred
 
@@ -21,8 +21,9 @@ class MiddlewareManager:
 
     def __init__(self, *middlewares):
         self.middlewares = middlewares
-        # Optional because process_spider_output and process_spider_exception can be None
-        self.methods: Dict[str, Deque[Optional[Callable]]] = defaultdict(deque)
+        # Only process_spider_output and process_spider_exception can be None.
+        # Only process_spider_output can be a tuple, and only until _async compatibility methods are removed.
+        self.methods: Dict[str, Deque[Union[None, Callable, Tuple[Callable, Callable]]]] = defaultdict(deque)
         for mw in middlewares:
             self._add_middleware(mw)
 

scrapy/spidermiddlewares/depth.py

@@ -7,7 +7,6 @@ See documentation in docs/topics/spider-middleware.rst
 import logging
 
 from scrapy.http import Request
-from scrapy.utils.asyncgen import _process_iterable_universal
 
 logger = logging.getLogger(__name__)
 
@@ -29,36 +28,42 @@ class DepthMiddleware:
         return cls(maxdepth, crawler.stats, verbose, prio)
 
     def process_spider_output(self, response, result, spider):
-        def _filter(request):
-            if isinstance(request, Request):
-                depth = response.meta['depth'] + 1
-                request.meta['depth'] = depth
-                if self.prio:
-                    request.priority -= depth * self.prio
-                if self.maxdepth and depth > self.maxdepth:
-                    logger.debug(
-                        "Ignoring link (depth > %(maxdepth)d): %(requrl)s ",
-                        {'maxdepth': self.maxdepth, 'requrl': request.url},
-                        extra={'spider': spider}
-                    )
-                    return False
-                else:
-                    if self.verbose_stats:
-                        self.stats.inc_value(f'request_depth_count/{depth}',
-                                             spider=spider)
-                    self.stats.max_value('request_depth_max', depth,
-                                         spider=spider)
+        # base case (depth=0)
+        if 'depth' not in response.meta:
+            response.meta['depth'] = 0
+            if self.verbose_stats:
+                self.stats.inc_value('request_depth_count/0', spider=spider)
+
+        return (r for r in result or () if self._filter(r, response, spider))
+
+    async def process_spider_output_async(self, response, result, spider):
+        # base case (depth=0)
+        if 'depth' not in response.meta:
+            response.meta['depth'] = 0
+            if self.verbose_stats:
+                self.stats.inc_value('request_depth_count/0', spider=spider)
+
+        async for r in result or ():
+            if self._filter(r, response, spider):
+                yield r
+
+    def _filter(self, request, response, spider):
+        if not isinstance(request, Request):
             return True
-
-        @_process_iterable_universal
-        async def process(result):
-            # base case (depth=0)
-            if 'depth' not in response.meta:
-                response.meta['depth'] = 0
-                if self.verbose_stats:
-                    self.stats.inc_value('request_depth_count/0', spider=spider)
-
-            async for r in result or ():
-                if _filter(r):
-                    yield r
-        return process(result)
+        depth = response.meta['depth'] + 1
+        request.meta['depth'] = depth
+        if self.prio:
+            request.priority -= depth * self.prio
+        if self.maxdepth and depth > self.maxdepth:
+            logger.debug(
+                "Ignoring link (depth > %(maxdepth)d): %(requrl)s ",
+                {'maxdepth': self.maxdepth, 'requrl': request.url},
+                extra={'spider': spider}
+            )
+            return False
+        if self.verbose_stats:
+            self.stats.inc_value(f'request_depth_count/{depth}',
+                                 spider=spider)
+        self.stats.max_value('request_depth_max', depth,
+                             spider=spider)
+        return True

scrapy/spidermiddlewares/offsite.py

@@ -9,7 +9,6 @@ import warnings
 
 from scrapy import signals
 from scrapy.http import Request
-from scrapy.utils.asyncgen import _process_iterable_universal
 from scrapy.utils.httpobj import urlparse_cached
 
 logger = logging.getLogger(__name__)
@@ -27,24 +26,27 @@ class OffsiteMiddleware:
         return o
 
     def process_spider_output(self, response, result, spider):
-        @_process_iterable_universal
-        async def process(result):
-            async for x in result:
-                if isinstance(x, Request):
-                    if x.dont_filter or self.should_follow(x, spider):
-                        yield x
-                    else:
-                        domain = urlparse_cached(x).hostname
-                        if domain and domain not in self.domains_seen:
-                            self.domains_seen.add(domain)
-                            logger.debug(
-                                "Filtered offsite request to %(domain)r: %(request)s",
-                                {'domain': domain, 'request': x}, extra={'spider': spider})
-                            self.stats.inc_value('offsite/domains', spider=spider)
-                        self.stats.inc_value('offsite/filtered', spider=spider)
-                else:
-                    yield x
-        return process(result)
+        return (r for r in result or () if self._filter(r, spider))
+
+    async def process_spider_output_async(self, response, result, spider):
+        async for r in result or ():
+            if self._filter(r, spider):
+                yield r
+
+    def _filter(self, request, spider) -> bool:
+        if not isinstance(request, Request):
+            return True
+        if request.dont_filter or self.should_follow(request, spider):
+            return True
+        domain = urlparse_cached(request).hostname
+        if domain and domain not in self.domains_seen:
+            self.domains_seen.add(domain)
+            logger.debug(
+                "Filtered offsite request to %(domain)r: %(request)s",
+                {'domain': domain, 'request': request}, extra={'spider': spider})
+            self.stats.inc_value('offsite/domains', spider=spider)
+        self.stats.inc_value('offsite/filtered', spider=spider)
+        return False
 
     def should_follow(self, request, spider):
         regex = self.host_regex

scrapy/spidermiddlewares/referer.py

@@ -11,7 +11,6 @@ from w3lib.url import safe_url_string
 from scrapy import signals
 from scrapy.exceptions import NotConfigured
 from scrapy.http import Request, Response
-from scrapy.utils.asyncgen import _process_iterable_universal
 from scrapy.utils.misc import load_object
 from scrapy.utils.python import to_unicode
 from scrapy.utils.url import strip_url
@@ -334,19 +333,18 @@ class RefererMiddleware:
         return cls() if cls else self.default_policy()
 
     def process_spider_output(self, response, result, spider):
-        def _set_referer(r):
-            if isinstance(r, Request):
-                referrer = self.policy(response, r).referrer(response.url, r.url)
-                if referrer is not None:
-                    r.headers.setdefault('Referer', referrer)
-            return r
+        return (self._set_referer(r, response) for r in result or ())
 
-        @_process_iterable_universal
-        async def process(result):
-            async for r in result or ():
-                yield _set_referer(r)
+    async def process_spider_output_async(self, response, result, spider):
+        async for r in result or ():
+            yield self._set_referer(r, response)
 
-        return process(result)
+    def _set_referer(self, r, response):
+        if isinstance(r, Request):
+            referrer = self.policy(response, r).referrer(response.url, r.url)
+            if referrer is not None:
+                r.headers.setdefault('Referer', referrer)
+        return r
 
     def request_scheduled(self, request, spider):
         # check redirected request to patch "Referer" header if necessary

scrapy/spidermiddlewares/urllength.py

@@ -8,7 +8,6 @@ import logging
 
 from scrapy.http import Request
 from scrapy.exceptions import NotConfigured
-from scrapy.utils.asyncgen import _process_iterable_universal
 
 logger = logging.getLogger(__name__)
 
@@ -26,22 +25,20 @@ class UrlLengthMiddleware:
         return cls(maxlength)
 
     def process_spider_output(self, response, result, spider):
-        def _filter(request):
-            if isinstance(request, Request) and len(request.url) > self.maxlength:
-                logger.info(
-                    "Ignoring link (url length > %(maxlength)d): %(url)s ",
-                    {'maxlength': self.maxlength, 'url': request.url},
-                    extra={'spider': spider}
-                )
-                spider.crawler.stats.inc_value('urllength/request_ignored_count', spider=spider)
-                return False
-            else:
-                return True
+        return (r for r in result or () if self._filter(r, spider))
 
-        @_process_iterable_universal
-        async def process(result):
-            async for r in result or ():
-                if _filter(r):
-                    yield r
+    async def process_spider_output_async(self, response, result, spider):
+        async for r in result or ():
+            if self._filter(r, spider):
+                yield r
 
-        return process(result)
+    def _filter(self, request, spider):
+        if isinstance(request, Request) and len(request.url) > self.maxlength:
+            logger.info(
+                "Ignoring link (url length > %(maxlength)d): %(url)s ",
+                {'maxlength': self.maxlength, 'url': request.url},
+                extra={'spider': spider}
+            )
+            spider.crawler.stats.inc_value('urllength/request_ignored_count', spider=spider)
+            return False
+        return True

scrapy/utils/asyncgen.py

@@ -1,6 +1,4 @@
-import functools
-import inspect
-from typing import AsyncGenerator, AsyncIterable, Callable, Generator, Iterable, Union
+from typing import AsyncGenerator, AsyncIterable, Iterable, Union
 
 
 async def collect_asyncgen(result: AsyncIterable):
@@ -18,46 +16,3 @@ async def as_async_generator(it: Union[Iterable, AsyncIterable]) -> AsyncGenerat
     else:
         for r in it:
             yield r
-
-
-# https://stackoverflow.com/a/66170760/113586
-def _process_iterable_universal(process_async: Callable):
-    """ Takes a function that takes an async iterable, args and kwargs. Returns
-    a function that takes any iterable, args and kwargs.
-
-    Requires that process_async only awaits on the iterable and synchronous functions,
-    so it's better to use this only in the Scrapy code itself.
-    """
-
-    # If this stops working, all internal uses can be just replaced with manually-written
-    # process_sync functions.
-
-    def process_sync(iterable: Iterable, *args, **kwargs) -> Generator:
-        agen = process_async(as_async_generator(iterable), *args, **kwargs)
-        if not inspect.isasyncgen(agen):
-            raise ValueError(f"process_async returned wrong type {type(agen)}")
-        sent = None
-        while True:
-            try:
-                gen = agen.asend(sent)
-                gen.send(None)
-            except StopIteration as e:
-                sent = yield e.value
-            except StopAsyncIteration:
-                return
-            else:
-                gen.throw(RuntimeError,
-                          f"Synchronously-called function '{process_async.__name__}' has blocked, "
-                          f"you can't use {_process_iterable_universal.__name__} with it.")
-
-    @functools.wraps(process_async)
-    def process(iterable: Union[Iterable, AsyncIterable], *args, **kwargs) -> Union[Generator, AsyncGenerator]:
-        if isinstance(iterable, AsyncIterable):
-            # call process_async directly
-            return process_async(iterable, *args, **kwargs)
-        if isinstance(iterable, Iterable):
-            # convert process_async to process_sync
-            return process_sync(iterable, *args, **kwargs)
-        raise TypeError(f"Wrong iterable type {type(iterable)}")
-
-    return process

scrapy/utils/python.py

@@ -10,7 +10,7 @@ import warnings
 import weakref
 from functools import partial, wraps
 from itertools import chain
-from typing import AsyncIterable, Iterable, Union
+from typing import AsyncGenerator, AsyncIterable, Iterable, Union
 
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.utils.asyncgen import as_async_generator
@@ -359,7 +359,7 @@ class MutableChain(Iterable):
         return self.__next__()
 
 
-async def _async_chain(*iterables: Union[Iterable, AsyncIterable]):
+async def _async_chain(*iterables: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
     for it in iterables:
         async for o in as_async_generator(it):
             yield o

tests/test_spidermiddleware.py

@@ -1,6 +1,8 @@
 import collections.abc
+from typing import Optional
 from unittest import mock
 
+from testfixtures import LogCapture
 from twisted.internet import defer
 from twisted.trial.unittest import TestCase
 from twisted.python.failure import Failure
@@ -8,8 +10,8 @@ from twisted.python.failure import Failure
 from scrapy.spiders import Spider
 from scrapy.http import Request, Response
 from scrapy.exceptions import _InvalidOutput
-from scrapy.utils.asyncgen import _process_iterable_universal, as_async_generator, collect_asyncgen
-from scrapy.utils.defer import deferred_from_coro
+from scrapy.utils.asyncgen import collect_asyncgen
+from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
 from scrapy.utils.test import get_crawler
 from scrapy.core.spidermw import SpiderMiddlewareManager
 
@@ -115,33 +117,40 @@ class BaseAsyncSpiderMiddlewareTestCase(SpiderMiddlewareTestCase):
 
     RESULT_COUNT = 3  # to simplify checks, let everything return 3 objects
 
+    @staticmethod
+    def _construct_mw_setting(*mw_classes, start_index: Optional[int] = None):
+        if start_index is None:
+            start_index = 10
+        return {i: c for c, i in enumerate(mw_classes, start=start_index)}
+
     @defer.inlineCallbacks
-    def _get_middleware_result(self, *mw_classes):
-        for mw_cls in mw_classes:
-            self.mwman._add_middleware(mw_cls())
+    def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
+        setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
+        self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES_BASE': {}, 'SPIDER_MIDDLEWARES': setting})
+        self.spider = self.crawler._create_spider('foo')
+        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
         result = yield self.mwman.scrape_response(self._scrape_func, self.response, self.request, self.spider)
         return result
 
     @defer.inlineCallbacks
-    def _test_simple_base(self, *mw_classes):
-        result = yield self._get_middleware_result(*mw_classes)
+    def _test_simple_base(self, *mw_classes, downgrade: bool = False, start_index: Optional[int] = None):
+        with LogCapture() as log:
+            result = yield self._get_middleware_result(*mw_classes, start_index=start_index)
         self.assertIsInstance(result, collections.abc.Iterable)
         result_list = list(result)
         self.assertEqual(len(result_list), self.RESULT_COUNT)
         self.assertIsInstance(result_list[0], self.ITEM_TYPE)
+        self.assertEqual("downgraded to a non-async" in str(log), downgrade)
 
     @defer.inlineCallbacks
-    def _test_asyncgen_base(self, *mw_classes):
-        result = yield self._get_middleware_result(*mw_classes)
+    def _test_asyncgen_base(self, *mw_classes, downgrade: bool = False, start_index: Optional[int] = None):
+        with LogCapture() as log:
+            result = yield self._get_middleware_result(*mw_classes, start_index=start_index)
         self.assertIsInstance(result, collections.abc.AsyncIterator)
         result_list = yield deferred_from_coro(collect_asyncgen(result))
         self.assertEqual(len(result_list), self.RESULT_COUNT)
         self.assertIsInstance(result_list[0], self.ITEM_TYPE)
-
-    @defer.inlineCallbacks
-    def _test_asyncgen_fail(self, *mw_classes):
-        with self.assertRaisesRegex(TypeError, "Synchronous .+ called with an async iterable"):
-            yield self._get_middleware_result(*mw_classes)
+        self.assertEqual("downgraded to a non-async" in str(log), downgrade)
 
 
 class ProcessSpiderOutputSimpleMiddleware:
@@ -152,17 +161,36 @@ class ProcessSpiderOutputSimpleMiddleware:
 
 class ProcessSpiderOutputAsyncGenMiddleware:
     async def process_spider_output(self, response, result, spider):
-        async for r in as_async_generator(result):
+        async for r in result:
             yield r
 
 
 class ProcessSpiderOutputUniversalMiddleware:
     def process_spider_output(self, response, result, spider):
-        @_process_iterable_universal
-        async def process(result):
-            async for r in result:
-                yield r
-        return process(result)
+        for r in result:
+            yield r
+
+    async def process_spider_output_async(self, response, result, spider):
+        async for r in result:
+            yield r
+
+
+class ProcessSpiderExceptionSimpleIterableMiddleware:
+    def process_spider_exception(self, response, exception, spider):
+        yield {'foo': 1}
+        yield {'foo': 2}
+        yield {'foo': 3}
+
+
+class ProcessSpiderExceptionAsyncIterableMiddleware:
+    async def process_spider_exception(self, response, exception, spider):
+        yield {'foo': 1}
+        d = defer.Deferred()
+        from twisted.internet import reactor
+        reactor.callLater(0, d.callback, None)
+        await maybe_deferred_to_future(d)
+        yield {'foo': 2}
+        yield {'foo': 3}
 
 
 class ProcessSpiderOutputSimple(BaseAsyncSpiderMiddlewareTestCase):
@@ -183,18 +211,19 @@ class ProcessSpiderOutputSimple(BaseAsyncSpiderMiddlewareTestCase):
         return self._test_simple_base(self.MW_SIMPLE)
 
     def test_asyncgen(self):
-        """ Asyncgen mw """
+        """ Asyncgen mw; upgrade """
         return self._test_asyncgen_base(self.MW_ASYNCGEN)
 
     def test_simple_asyncgen(self):
-        """ Simple mw -> asyncgen mw """
+        """ Simple mw -> asyncgen mw; upgrade """
         return self._test_asyncgen_base(self.MW_ASYNCGEN,
                                         self.MW_SIMPLE)
 
     def test_asyncgen_simple(self):
-        """ Asyncgen mw -> simple mw; cannot work """
-        return self._test_asyncgen_fail(self.MW_SIMPLE,
-                                        self.MW_ASYNCGEN)
+        """ Asyncgen mw -> simple mw; upgrade then downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_ASYNCGEN,
+                                      downgrade=True)
 
     def test_universal(self):
         """ Universal mw """
@@ -211,12 +240,12 @@ class ProcessSpiderOutputSimple(BaseAsyncSpiderMiddlewareTestCase):
                                       self.MW_SIMPLE)
 
     def test_universal_asyncgen(self):
-        """ Universal mw -> asyncgen mw """
+        """ Universal mw -> asyncgen mw; upgrade """
         return self._test_asyncgen_base(self.MW_ASYNCGEN,
                                         self.MW_UNIVERSAL)
 
     def test_asyncgen_universal(self):
-        """ Asyncgen mw -> universal mw """
+        """ Asyncgen mw -> universal mw; upgrade """
         return self._test_asyncgen_base(self.MW_UNIVERSAL,
                                         self.MW_ASYNCGEN)
 
@@ -229,27 +258,31 @@ class ProcessSpiderOutputAsyncGen(ProcessSpiderOutputSimple):
             yield item
 
     def test_simple(self):
-        """ Simple mw; cannot work """
-        return self._test_asyncgen_fail(self.MW_SIMPLE)
+        """ Simple mw; downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      downgrade=True)
 
     def test_simple_asyncgen(self):
-        """ Simple mw -> asyncgen mw; cannot work """
-        return self._test_asyncgen_fail(self.MW_ASYNCGEN,
-                                        self.MW_SIMPLE)
+        """ Simple mw -> asyncgen mw; downgrade then upgrade """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN,
+                                        self.MW_SIMPLE,
+                                        downgrade=True)
 
     def test_universal(self):
         """ Universal mw """
         return self._test_asyncgen_base(self.MW_UNIVERSAL)
 
     def test_universal_simple(self):
-        """ Universal mw -> simple mw; cannot work """
-        return self._test_asyncgen_fail(self.MW_SIMPLE,
-                                        self.MW_UNIVERSAL)
+        """ Universal mw -> simple mw; downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_UNIVERSAL,
+                                      downgrade=True)
 
     def test_simple_universal(self):
-        """ Simple mw -> universal mw; cannot work """
-        return self._test_asyncgen_fail(self.MW_UNIVERSAL,
-                                        self.MW_SIMPLE)
+        """ Simple mw -> universal mw; downgrade """
+        return self._test_simple_base(self.MW_UNIVERSAL,
+                                      self.MW_SIMPLE,
+                                      downgrade=True)
 
 
 class ProcessStartRequestsSimpleMiddleware:
@@ -269,13 +302,198 @@ class ProcessStartRequestsSimple(BaseAsyncSpiderMiddlewareTestCase):
             yield Request(f'https://example.com/{i}', dont_filter=True)
 
     @defer.inlineCallbacks
-    def _get_middleware_result(self, *mw_classes):
-        for mw_cls in mw_classes:
-            self.mwman._add_middleware(mw_cls())
+    def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
+        setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
+        self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES_BASE': {}, 'SPIDER_MIDDLEWARES': setting})
+        self.spider = self.crawler._create_spider('foo')
+        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
         start_requests = iter(self._start_requests())
         results = yield self.mwman.process_start_requests(start_requests, self.spider)
         return results
 
     def test_simple(self):
         """ Simple mw """
-        self._test_simple_base(self.MW_SIMPLE)
+        return self._test_simple_base(self.MW_SIMPLE)
+
+
+class UniversalMiddlewareNoSync:
+    async def process_spider_output_async(self, response, result, spider):
+        yield
+
+
+class UniversalMiddlewareBothSync:
+    def process_spider_output(self, response, result, spider):
+        yield
+
+    def process_spider_output_async(self, response, result, spider):
+        yield
+
+
+class UniversalMiddlewareBothAsync:
+    async def process_spider_output(self, response, result, spider):
+        yield
+
+    async def process_spider_output_async(self, response, result, spider):
+        yield
+
+
+class UniversalMiddlewareManagerTest(TestCase):
+    def setUp(self):
+        self.mwman = SpiderMiddlewareManager()
+
+    def test_simple_mw(self):
+        mw = ProcessSpiderOutputSimpleMiddleware
+        self.mwman._add_middleware(mw)
+        self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
+
+    def test_async_mw(self):
+        mw = ProcessSpiderOutputAsyncGenMiddleware
+        self.mwman._add_middleware(mw)
+        self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
+
+    def test_universal_mw(self):
+        mw = ProcessSpiderOutputUniversalMiddleware
+        self.mwman._add_middleware(mw)
+        self.assertEqual(self.mwman.methods['process_spider_output'][0],
+                         (mw.process_spider_output, mw.process_spider_output_async))
+
+    def test_universal_mw_no_sync(self):
+        with LogCapture() as log:
+            self.mwman._add_middleware(UniversalMiddlewareNoSync)
+        self.assertIn("UniversalMiddlewareNoSync has process_spider_output_async"
+                      " without process_spider_output", str(log))
+        self.assertEqual(self.mwman.methods['process_spider_output'][0], None)
+
+    def test_universal_mw_both_sync(self):
+        mw = UniversalMiddlewareBothSync
+        with LogCapture() as log:
+            self.mwman._add_middleware(mw)
+        self.assertIn("UniversalMiddlewareBothSync.process_spider_output_async "
+                      "is not an async generator function", str(log))
+        self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
+
+    def test_universal_mw_both_async(self):
+        with LogCapture() as log:
+            self.mwman._add_middleware(UniversalMiddlewareBothAsync)
+        self.assertIn("UniversalMiddlewareBothAsync.process_spider_output "
+                      "is an async generator function while process_spider_output_async exists",
+                      str(log))
+        self.assertEqual(self.mwman.methods['process_spider_output'][0], None)
+
+
+class BuiltinMiddlewareSimpleTest(BaseAsyncSpiderMiddlewareTestCase):
+    ITEM_TYPE = dict
+    MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
+    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
+    MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
+
+    def _scrape_func(self, *args, **kwargs):
+        yield {'foo': 1}
+        yield {'foo': 2}
+        yield {'foo': 3}
+
+    @defer.inlineCallbacks
+    def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
+        setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
+        self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES': setting})
+        self.spider = self.crawler._create_spider('foo')
+        self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
+        result = yield self.mwman.scrape_response(self._scrape_func, self.response, self.request, self.spider)
+        return result
+
+    def test_just_builtin(self):
+        return self._test_simple_base()
+
+    def test_builtin_simple(self):
+        return self._test_simple_base(self.MW_SIMPLE, start_index=1000)
+
+    def test_builtin_async(self):
+        """ Upgrade """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN, start_index=1000)
+
+    def test_builtin_universal(self):
+        return self._test_simple_base(self.MW_UNIVERSAL, start_index=1000)
+
+    def test_simple_builtin(self):
+        return self._test_simple_base(self.MW_SIMPLE)
+
+    def test_async_builtin(self):
+        """ Upgrade """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN)
+
+    def test_universal_builtin(self):
+        return self._test_simple_base(self.MW_UNIVERSAL)
+
+
+class BuiltinMiddlewareAsyncGenTest(BuiltinMiddlewareSimpleTest):
+    async def _scrape_func(self, *args, **kwargs):
+        for item in super()._scrape_func():
+            yield item
+
+    def test_just_builtin(self):
+        return self._test_asyncgen_base()
+
+    def test_builtin_simple(self):
+        """ Downgrade """
+        return self._test_simple_base(self.MW_SIMPLE, downgrade=True, start_index=1000)
+
+    def test_builtin_async(self):
+        return self._test_asyncgen_base(self.MW_ASYNCGEN, start_index=1000)
+
+    def test_builtin_universal(self):
+        return self._test_asyncgen_base(self.MW_UNIVERSAL, start_index=1000)
+
+    def test_simple_builtin(self):
+        """ Downgrade """
+        return self._test_simple_base(self.MW_SIMPLE, downgrade=True)
+
+    def test_async_builtin(self):
+        return self._test_asyncgen_base(self.MW_ASYNCGEN)
+
+    def test_universal_builtin(self):
+        return self._test_asyncgen_base(self.MW_UNIVERSAL)
+
+
+class ProcessSpiderExceptionTest(BaseAsyncSpiderMiddlewareTestCase):
+    ITEM_TYPE = dict
+    MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
+    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
+    MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
+    MW_EXC_SIMPLE = ProcessSpiderExceptionSimpleIterableMiddleware
+    MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIterableMiddleware
+
+    def _scrape_func(self, *args, **kwargs):
+        1 / 0
+
+    @defer.inlineCallbacks
+    def _test_asyncgen_nodowngrade(self, *mw_classes):
+        with self.assertRaisesRegex(_InvalidOutput, "Async iterable returned from .+ cannot be downgraded"):
+            yield self._get_middleware_result(*mw_classes)
+
+    def test_exc_simple(self):
+        """ Simple exc mw """
+        return self._test_simple_base(self.MW_EXC_SIMPLE)
+
+    def test_exc_async(self):
+        """ Async exc mw """
+        return self._test_asyncgen_base(self.MW_EXC_ASYNCGEN)
+
+    def test_exc_simple_simple(self):
+        """ Simple exc mw -> simple output mw """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_EXC_SIMPLE)
+
+    def test_exc_async_async(self):
+        """ Async exc mw -> async output mw """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN,
+                                        self.MW_EXC_ASYNCGEN)
+
+    def test_exc_simple_async(self):
+        """ Simple exc mw -> async output mw; upgrade """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN,
+                                        self.MW_EXC_SIMPLE)
+
+    def test_exc_async_simple(self):
+        """ Async exc mw -> simple output mw; cannot work as downgrading is not supported """
+        return self._test_asyncgen_nodowngrade(self.MW_SIMPLE,
+                                               self.MW_EXC_ASYNCGEN)

tests/test_spidermiddleware_output_chain.py

@@ -28,6 +28,7 @@ class RecoveryMiddleware:
 class RecoverySpider(Spider):
     name = 'RecoverySpider'
     custom_settings = {
+        'SPIDER_MIDDLEWARES_BASE': {},
         'SPIDER_MIDDLEWARES': {
             RecoveryMiddleware: 10,
         },
@@ -107,6 +108,13 @@ class GeneratorCallbackSpider(Spider):
         raise ImportError()
 
 
+class AsyncGeneratorCallbackSpider(GeneratorCallbackSpider):
+    async def parse(self, response):
+        yield {'test': 1}
+        yield {'test': 2}
+        raise ImportError()
+
+
 # ================================================================================
 # (2.1) exceptions from a spider callback (generator, middleware right after callback)
 class GeneratorCallbackSpiderMiddlewareRightAfterSpider(GeneratorCallbackSpider):
@@ -360,6 +368,15 @@ class TestSpiderMiddleware(TestCase):
         self.assertIn("Middleware: ImportError exception caught", str(log2))
         self.assertIn("'item_scraped_count': 2", str(log2))
 
+    @defer.inlineCallbacks
+    def test_async_generator_callback(self):
+        """
+        Same as test_generator_callback but with an async callback.
+        """
+        log2 = yield self.crawl_log(AsyncGeneratorCallbackSpider)
+        self.assertIn("Middleware: ImportError exception caught", str(log2))
+        self.assertIn("'item_scraped_count': 2", str(log2))
+
     @defer.inlineCallbacks
     def test_generator_callback_right_after_callback(self):
         """

tests/test_utils_asyncgen.py

@@ -1,7 +1,6 @@
-from twisted.internet.defer import Deferred
 from twisted.trial import unittest
 
-from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen, _process_iterable_universal
+from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.defer import deferred_f_from_coro_f
 
 
@@ -19,52 +18,3 @@ class AsyncgenUtilsTest(unittest.TestCase):
         ag = as_async_generator(range(42))
         results = await collect_asyncgen(ag)
         self.assertEqual(results, list(range(42)))
-
-
-@_process_iterable_universal
-async def process_iterable(iterable):
-    async for i in iterable:
-        yield i * 2
-
-
-@_process_iterable_universal
-async def process_iterable_awaiting(iterable):
-    async for i in iterable:
-        yield i * 2
-        d = Deferred()
-        from twisted.internet import reactor
-        reactor.callLater(0, d.callback, 42)
-        await d
-
-
-class ProcessIterableUniversalTest(unittest.TestCase):
-
-    def test_normal(self):
-        iterable = iter([1, 2, 3])
-        results = list(process_iterable(iterable))
-        self.assertEqual(results, [2, 4, 6])
-
-    @deferred_f_from_coro_f
-    async def test_async(self):
-        iterable = as_async_generator([1, 2, 3])
-        results = await collect_asyncgen(process_iterable(iterable))
-        self.assertEqual(results, [2, 4, 6])
-
-    @deferred_f_from_coro_f
-    async def test_blocking(self):
-        iterable = [1, 2, 3]
-        with self.assertRaisesRegex(RuntimeError, "Synchronously-called function"):
-            list(process_iterable_awaiting(iterable))
-
-    def test_invalid_iterable(self):
-        with self.assertRaisesRegex(TypeError, "Wrong iterable type"):
-            process_iterable(None)
-
-    @deferred_f_from_coro_f
-    async def test_invalid_process(self):
-        @_process_iterable_universal
-        def process_iterable_invalid(iterable):
-            pass
-
-        with self.assertRaisesRegex(ValueError, "process_async returned wrong type"):
-            list(process_iterable_invalid([]))