Repository: scrapy/scrapy
Pull request #4978 (2021-02-05 14:22:50)

Support for async callbacks
This PR adds support for defining `async def parse_*` and using both `yield` and `await` inside, with or without the asyncio reactor. There is one caveat: if the asyncio reactor is enabled, you cannot await on Deferreds directly and need to wrap them. I've added this to docs.

This change was tested on two real-world projects, for one of them I also replaced using `inline_requests` to get HTTP responses inside a callback with alternatively `treq` or `aiohttp`. I didn't test the performance, but as far as I can see the existing non-async callbacks shouldn't have any significant code path changes so there should be no regressions and if there are any perf problems with async callbacks we can solve that later.

This doesn't require or support async def start_requests.

My main concern with this is large and non-intuitive `scrapy.utils.defer._AsyncCooperatorAdapter`. I would like to replace it with something better if that's possible but for now it seems to work correctly. I've tried to document it as comprehensively as possible.

I've split it in several commits with different features in case this help with review for someone.


Patch:
diff --git a/docs/index.rst b/docs/index.rst
index 40c6cb485..5404969e0 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -225,10 +225,11 @@ Extending Scrapy
    topics/downloader-middleware
    topics/spider-middleware
    topics/extensions
-   topics/api
    topics/signals
    topics/scheduler
    topics/exporters
+   topics/components
+   topics/api
 
 
 :doc:`topics/architecture`
@@ -243,9 +244,6 @@ Extending Scrapy
 :doc:`topics/extensions`
     Extend Scrapy with your custom functionality
 
-:doc:`topics/api`
-    Use it on extensions and middlewares to extend Scrapy functionality
-
 :doc:`topics/signals`
     See all available signals and how to work with them.
 
@@ -255,6 +253,13 @@ Extending Scrapy
 :doc:`topics/exporters`
     Quickly export your scraped items to a file (XML, CSV, etc).
 
+:doc:`topics/components`
+    Learn the common API and some good practices when building custom Scrapy
+    components.
+
+:doc:`topics/api`
+    Use it on extensions and middlewares to extend Scrapy functionality.
+
 
 All the rest
 ============
diff --git a/docs/topics/asyncio.rst b/docs/topics/asyncio.rst
index 3a6941a2c..dbee7146d 100644
--- a/docs/topics/asyncio.rst
+++ b/docs/topics/asyncio.rst
@@ -96,3 +96,27 @@ Futures. Scrapy provides two helpers for this:
          down to Scrapy 2.0 (earlier versions do not support
          :mod:`asyncio`), you can copy the implementation of these functions
          into your own code.
+
+
+.. _enforce-asyncio-requirement:
+
+Enforcing asyncio as a requirement
+==================================
+
+If you are writing a :ref:`component <topics-components>` that requires asyncio
+to work, use :func:`scrapy.utils.reactor.is_asyncio_reactor_installed` to
+:ref:`enforce it as a requirement <enforce-component-requirements>`. For
+example::
+
+    from scrapy.utils.reactor import is_asyncio_reactor_installed
+
+    class MyComponent:
+
+        def __init__(self):
+            if not is_asyncio_reactor_installed():
+                raise ValueError(
+                    f"{MyComponent.__qualname__} requires the asyncio Twisted "
+                    f"reactor. Make sure you have it configured in the "
+                    f"TWISTED_REACTOR setting. See the asyncio documentation "
+                    f"of Scrapy for more information."
+                )
diff --git a/docs/topics/components.rst b/docs/topics/components.rst
new file mode 100644
index 000000000..c44f3def2
--- /dev/null
+++ b/docs/topics/components.rst
@@ -0,0 +1,84 @@
+.. _topics-components:
+
+==========
+Components
+==========
+
+A Scrapy component is any class whose objects are created using
+:func:`scrapy.utils.misc.create_instance`.
+
+That includes the classes that you may assign to the following settings:
+
+-   :setting:`DNS_RESOLVER`
+
+-   :setting:`DOWNLOAD_HANDLERS`
+
+-   :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY`
+
+-   :setting:`DOWNLOADER_MIDDLEWARES`
+
+-   :setting:`DUPEFILTER_CLASS`
+
+-   :setting:`EXTENSIONS`
+
+-   :setting:`FEED_EXPORTERS`
+
+-   :setting:`FEED_STORAGES`
+
+-   :setting:`ITEM_PIPELINES`
+
+-   :setting:`SCHEDULER`
+
+-   :setting:`SCHEDULER_DISK_QUEUE`
+
+-   :setting:`SCHEDULER_MEMORY_QUEUE`
+
+-   :setting:`SCHEDULER_PRIORITY_QUEUE`
+
+-   :setting:`SPIDER_MIDDLEWARES`
+
+Third-party Scrapy components may also let you define additional Scrapy
+components, usually configurable through :ref:`settings <topics-settings>`, to
+modify their behavior.
+
+.. _enforce-component-requirements:
+
+Enforcing component requirements
+================================
+
+Sometimes, your components may only be intended to work under certain
+conditions. For example, they may require a minimum version of Scrapy to work as
+intended, or they may require certain settings to have specific values.
+
+In addition to describing those conditions in the documentation of your
+component, it is a good practice to raise an exception from the ``__init__``
+method of your component if those conditions are not met at run time.
+
+In the case of :ref:`downloader middlewares <topics-downloader-middleware>`,
+:ref:`extensions <topics-extensions>`, :ref:`item pipelines
+<topics-item-pipeline>`, and :ref:`spider middlewares
+<topics-spider-middleware>`, you should raise
+:exc:`scrapy.exceptions.NotConfigured`, passing a description of the issue as a
+parameter to the exception so that it is printed in the logs, for the user to
+see. For other components, feel free to raise whatever other exception feels
+right to you; for example, :exc:`RuntimeError` would make sense for a Scrapy
+version mismatch, while :exc:`ValueError` may be better if the issue is the
+value of a setting.
+
+If your requirement is a minimum Scrapy version, you may use
+:attr:`scrapy.__version__` to enforce your requirement. For example::
+
+    from pkg_resources import parse_version
+
+    import scrapy
+
+    class MyComponent:
+
+        def __init__(self):
+            if parse_version(scrapy.__version__) < parse_version('VERSION'):
+                raise RuntimeError(
+                    f"{MyComponent.__qualname__} requires Scrapy VERSION or "
+                    f"later, which allow defining the process_spider_output "
+                    f"method of spider middlewares as an asynchronous "
+                    f"generator."
+                )
diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst
index 549552bd1..750263385 100644
--- a/docs/topics/coroutines.rst
+++ b/docs/topics/coroutines.rst
@@ -19,14 +19,12 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
 
 -   :class:`~scrapy.Request` callbacks.
 
-    .. note:: The callback output is not processed until the whole callback
-        finishes.
+    If you are using any custom or third-party :ref:`spider middleware
+    <topics-spider-middleware>`, see :ref:`sync-async-spider-middleware`.
 
-        As a side effect, if the callback raises an exception, none of its
-        output is processed.
-
-        This is a known caveat of the current implementation that we aim to
-        address in a future version of Scrapy.
+    .. versionchanged:: VERSION
+       Output of async callbacks is now processed asynchronously instead of
+       collecting all of it first.
 
 -   The :meth:`process_item` method of
     :ref:`item pipelines <topics-item-pipeline>`.
@@ -41,12 +39,26 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
 
 -   :ref:`Signal handlers that support deferreds <signal-deferred>`.
 
-Usage
-=====
+-   The
+    :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output`
+    method of :ref:`spider middlewares <topics-spider-middleware>`.
+
+    It must be defined as an :term:`asynchronous generator`. The input
+    ``result`` parameter is an :term:`asynchronous iterable`.
+
+    See also :ref:`sync-async-spider-middleware` and
+    :ref:`universal-spider-middleware`.
 
-There are several use cases for coroutines in Scrapy. Code that would
-return Deferreds when written for previous Scrapy versions, such as downloader
-middlewares and signal handlers, can be rewritten to be shorter and cleaner::
+    .. versionadded:: VERSION
+
+General usage
+=============
+
+There are several use cases for coroutines in Scrapy.
+
+Code that would return Deferreds when written for previous Scrapy versions,
+such as downloader middlewares and signal handlers, can be rewritten to be
+shorter and cleaner::
 
     from itemadapter import ItemAdapter
 
@@ -106,7 +118,100 @@ 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
+
+
+.. _sync-async-spider-middleware:
+
+Mixing synchronous and asynchronous spider middlewares
+======================================================
+
+.. versionadded:: VERSION
+
+The output of a :class:`~scrapy.Request` callback is passed as the ``result``
+parameter to the
+:meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output` method
+of the first :ref:`spider middleware <topics-spider-middleware>` from the
+:ref:`list of active spider middlewares <topics-spider-middleware-setting>`.
+Then the output of that ``process_spider_output`` method is passed to the
+``process_spider_output`` method of the next spider middleware, and so on for
+every active spider middleware.
+
+Scrapy supports mixing :ref:`coroutine methods <async>` and synchronous methods
+in this chain of calls.
+
+However, if any of the ``process_spider_output`` methods is defined as a
+synchronous method, and the previous ``Request`` callback or
+``process_spider_output`` method is a coroutine, there are some drawbacks to
+the asynchronous-to-synchronous conversion that Scrapy does so that the
+synchronous ``process_spider_output`` method gets a synchronous iterable as its
+``result`` parameter:
+
+-   The whole output of the previous ``Request`` callback or
+    ``process_spider_output`` method is awaited at this point.
+
+-   If an exception raises while awaiting the output of the previous
+    ``Request`` callback or ``process_spider_output`` method, none of that
+    output will be processed.
+
+    This contrasts with the regular behavior, where all items yielded before
+    an exception raises are processed.
+
+Asynchronous-to-synchronous conversions are supported for backward
+compatibility, but they are deprecated and will stop working in a future
+version of Scrapy.
+
+To avoid asynchronous-to-synchronous conversions, when defining ``Request``
+callbacks as coroutine methods or when using spider middlewares whose
+``process_spider_output`` method is an :term:`asynchronous generator`, all
+active spider middlewares must either have their ``process_spider_output``
+method defined as an asynchronous generator or :ref:`define a
+process_spider_output_async method <universal-spider-middleware>`.
+
+.. note:: When using third-party spider middlewares that only define a
+          synchronous ``process_spider_output`` method, consider
+          :ref:`making them universal <universal-spider-middleware>` through
+          :ref:`subclassing <tut-inheritance>`.
+
+
+.. _universal-spider-middleware:
+
+Universal spider middlewares
+============================
+
+.. versionadded:: VERSION
+
+To allow writing a spider middleware that supports asynchronous execution of
+its ``process_spider_output`` method in Scrapy VERSION and later (avoiding
+:ref:`asynchronous-to-synchronous conversions <sync-async-spider-middleware>`)
+while maintaining support for older Scrapy versions, you may define
+``process_spider_output`` as a synchronous method and define an
+:term:`asynchronous generator` version of that method with an alternative name:
+``process_spider_output_async``.
+
+For example::
+
+    class UniversalSpiderMiddleware:
+        def process_spider_output(self, response, result, spider):
+            for r in result:
+                # ... do something with r
+                yield r
+
+        async def process_spider_output_async(self, response, result, spider):
+            async for r in result:
+                # ... do something with r
+                yield r
+
+.. note:: This is an interim measure to allow, for a time, to write code that
+          works in Scrapy VERSION and later without requiring
+          asynchronous-to-synchronous conversions, and works in earlier Scrapy
+          versions as well.
+
+          In some future version of Scrapy, however, this feature will be
+          deprecated and, eventually, in a later version of Scrapy, this
+          feature will be removed, and all spider middlewares will be expected
+          to define their ``process_spider_output`` method as an asynchronous
+          generator.
diff --git a/docs/topics/spider-middleware.rst b/docs/topics/spider-middleware.rst
index f27bc79c0..816cb5e03 100644
--- a/docs/topics/spider-middleware.rst
+++ b/docs/topics/spider-middleware.rst
@@ -102,27 +102,47 @@ object gives you access, for example, to the :ref:`settings <topics-settings>`.
         it has processed the response.
 
         :meth:`process_spider_output` must return an iterable of
-        :class:`~scrapy.Request` objects and :ref:`item object
+        :class:`~scrapy.Request` objects and :ref:`item objects
         <topics-items>`.
 
+        .. versionchanged:: VERSION
+           This method may be defined as an :term:`asynchronous generator`, in
+           which case ``result`` is an :term:`asynchronous iterable`.
+
+        Consider defining this method as an :term:`asynchronous generator`,
+        which will be a requirement in a future version of Scrapy. However, if
+        you plan on sharing your spider middleware with other people, consider
+        either :ref:`enforcing Scrapy VERSION <enforce-component-requirements>`
+        as a minimum requirement of your spider middleware, or :ref:`making
+        your spider middleware universal <universal-spider-middleware>` so that
+        it works with Scrapy versions earlier than Scrapy VERSION.
+
         :param response: the response which generated this output from the
           spider
         :type response: :class:`~scrapy.http.Response` object
 
         :param result: the result returned by the spider
         :type result: an iterable of :class:`~scrapy.Request` objects and
-          :ref:`item object <topics-items>`
+          :ref:`item objects <topics-items>`
 
         :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 defined, this method must be an :term:`asynchronous generator`,
+        which will be called instead of :meth:`process_spider_output` if
+        ``result`` is an :term:`asynchronous iterable`.
+
     .. method:: process_spider_exception(response, exception, spider)
 
         This method is called when a spider or :meth:`process_spider_output`
         method (from a previous spider middleware) raises an exception.
 
         :meth:`process_spider_exception` should return either ``None`` or an
-        iterable of :class:`~scrapy.Request` or :ref:`item <topics-items>` 
+        iterable of :class:`~scrapy.Request` or :ref:`item <topics-items>`
         objects.
 
         If it returns ``None``, Scrapy will continue processing this exception,
diff --git a/pylintrc b/pylintrc
index d8e47dc11..18819feba 100644
--- a/pylintrc
+++ b/pylintrc
@@ -47,6 +47,7 @@ disable=abstract-method,
         keyword-arg-before-vararg,
         line-too-long,
         logging-format-interpolation,
+        logging-fstring-interpolation,
         logging-not-lazy,
         lost-exception,
         method-hidden,
diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index f40bccbb3..e1fdd8d13 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -1,9 +1,8 @@
 """This module implements the Scraper component which parses responses and
 extracts information from them"""
-
 import logging
 from collections import deque
-from typing import Any, Deque, Iterable, Optional, Set, Tuple, Union
+from typing import Any, AsyncGenerator, AsyncIterable, Deque, Generator, Iterable, Optional, Set, Tuple, Union
 
 from itemadapter import is_item
 from twisted.internet.defer import Deferred, inlineCallbacks
@@ -13,7 +12,15 @@ from scrapy import signals, Spider
 from scrapy.core.spidermw import SpiderMiddlewareManager
 from scrapy.exceptions import CloseSpider, DropItem, IgnoreRequest
 from scrapy.http import Request, Response
-from scrapy.utils.defer import defer_fail, defer_succeed, iter_errback, parallel
+from scrapy.utils.defer import (
+    aiter_errback,
+    defer_fail,
+    defer_succeed,
+    iter_errback,
+    parallel,
+    parallel_async,
+)
+
 from scrapy.utils.log import failure_to_exc_info, logformatter_adapter
 from scrapy.utils.misc import load_object, warn_on_generator_with_return_value
 from scrapy.utils.spider import iterate_spider_output
@@ -185,12 +192,19 @@ 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 = iter_errback(result, self.handle_spider_error, request, response, spider)
-        dfd = parallel(it, self.concurrent_items, self._process_spidermw_output,
-                       request, response, spider)
+        it: Union[Generator, AsyncGenerator]
+        if isinstance(result, AsyncIterable):
+            it = aiter_errback(result, self.handle_spider_error, request, response, spider)
+            dfd = parallel_async(it, self.concurrent_items, self._process_spidermw_output,
+                                 request, response, spider)
+        else:
+            it = iter_errback(result, self.handle_spider_error, request, response, spider)
+            dfd = parallel(it, self.concurrent_items, self._process_spidermw_output,
+                           request, response, spider)
         return dfd
 
     def _process_spidermw_output(self, output: Any, request: Request, response: Response,
diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py
index 7cdc28284..1aa02f29f 100644
--- a/scrapy/core/spidermw.py
+++ b/scrapy/core/spidermw.py
@@ -3,32 +3,42 @@ Spider Middleware manager
 
 See documentation in docs/topics/spider-middleware.rst
 """
+import logging
+from inspect import isasyncgenfunction, iscoroutine
 from itertools import islice
-from typing import Any, 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 as_async_generator, collect_asyncgen
 from scrapy.utils.conf import build_component_list
-from scrapy.utils.defer import mustbe_deferred
-from scrapy.utils.python import MutableChain
+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]
 
 
 def _isiterable(o) -> bool:
-    return isinstance(o, Iterable)
+    return isinstance(o, (Iterable, AsyncIterable))
 
 
 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'))
@@ -39,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)
@@ -51,7 +61,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
             try:
                 result = method(response=response, spider=spider)
                 if result is not None:
-                    msg = (f"Middleware {method.__qualname__} must return None "
+                    msg = (f"{method.__qualname__} must return None "
                            f"or raise an exception, got {type(result)}")
                     raise _InvalidOutput(msg)
             except _InvalidOutput:
@@ -60,17 +70,35 @@ class SpiderMiddlewareManager(MiddlewareManager):
                 return scrape_func(Failure(), request, spider)
         return scrape_func(response, request, spider)
 
-    def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Iterable,
-                           exception_processor_index: int, recover_to: MutableChain) -> Generator:
-        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)
+    def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Union[Iterable, AsyncIterable],
+                           exception_processor_index: int, recover_to: Union[MutableChain, MutableAsyncChain]
+                           ) -> Union[Generator, AsyncGenerator]:
+
+        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
+            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)
+
+        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]:
@@ -82,30 +110,83 @@ 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:
-                msg = (f"Middleware {method.__qualname__} must return None "
+                msg = (f"{method.__qualname__} must return None "
                        f"or an iterable, got {type(result)}")
                 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: Iterable, start_index: int = 0) -> MutableChain:
+                               result: Union[Iterable, AsyncIterable], start_index: int = 0
+                               ) -> 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 = MutableChain()
+        recovered: Union[MutableChain, MutableAsyncChain]
+        last_result_is_async = isinstance(result, AsyncIterable)
+        if last_result_is_async:
+            recovered = MutableAsyncChain()
+        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:
@@ -116,28 +197,77 @@ class SpiderMiddlewareManager(MiddlewareManager):
             if _isiterable(result):
                 result = self._evaluate_iterable(response, spider, result, method_index + 1, recovered)
             else:
-                msg = (f"Middleware {method.__qualname__} must return an "
-                       f"iterable, got {type(result)}")
+                if iscoroutine(result):
+                    result.close()  # Silence warning about not awaiting
+                    msg = (
+                        f"{method.__qualname__} must be an asynchronous "
+                        f"generator (i.e. use yield)"
+                    )
+                else:
+                    msg = (
+                        f"{method.__qualname__} must return an iterable, got "
+                        f"{type(result)}"
+                    )
                 raise _InvalidOutput(msg)
+            last_result_is_async = isinstance(result, AsyncIterable)
 
-        return MutableChain(result, recovered)
+        if last_result_is_async:
+            return MutableAsyncChain(result, recovered)
+        else:
+            return MutableChain(result, recovered)  # type: ignore[arg-type]
 
-    def _process_callback_output(self, response: Response, spider: Spider, result: Iterable) -> MutableChain:
-        recovered = MutableChain()
+    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)
-        return MutableChain(self._process_spider_output(response, spider, result), recovered)
+        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: Iterable) -> MutableChain:
-            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
diff --git a/scrapy/middleware.py b/scrapy/middleware.py
index 2eb1d8609..8d7e5a602 100644
--- a/scrapy/middleware.py
+++ b/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)
 
diff --git a/scrapy/spidermiddlewares/depth.py b/scrapy/spidermiddlewares/depth.py
index 776a6879a..4c923b1b3 100644
--- a/scrapy/spidermiddlewares/depth.py
+++ b/scrapy/spidermiddlewares/depth.py
@@ -28,31 +28,39 @@ 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)
-            return True
+        self._init_depth(response, spider)
+        return (r for r in result or () if self._filter(r, response, spider))
+
+    async def process_spider_output_async(self, response, result, spider):
+        self._init_depth(response, spider)
+        async for r in result or ():
+            if self._filter(r, response, spider):
+                yield r
 
+    def _init_depth(self, response, 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 _filter(r))
+    def _filter(self, request, response, spider):
+        if not isinstance(request, Request):
+            return True
+        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
diff --git a/scrapy/spidermiddlewares/offsite.py b/scrapy/spidermiddlewares/offsite.py
index 6e4efda97..448bc1367 100644
--- a/scrapy/spidermiddlewares/offsite.py
+++ b/scrapy/spidermiddlewares/offsite.py
@@ -26,21 +26,27 @@ class OffsiteMiddleware:
         return o
 
     def process_spider_output(self, response, result, spider):
-        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 (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
diff --git a/scrapy/spidermiddlewares/referer.py b/scrapy/spidermiddlewares/referer.py
index 608c0eea5..8027beb92 100644
--- a/scrapy/spidermiddlewares/referer.py
+++ b/scrapy/spidermiddlewares/referer.py
@@ -333,13 +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 (_set_referer(r) for r in result or ())
+        return (self._set_referer(r, response) for r in result or ())
+
+    async def process_spider_output_async(self, response, result, spider):
+        async for r in result or ():
+            yield self._set_referer(r, response)
+
+    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
diff --git a/scrapy/spidermiddlewares/urllength.py b/scrapy/spidermiddlewares/urllength.py
index 450d4ff40..7ad64d2af 100644
--- a/scrapy/spidermiddlewares/urllength.py
+++ b/scrapy/spidermiddlewares/urllength.py
@@ -25,16 +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 _filter(r))
+        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):
+        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
diff --git a/scrapy/utils/asyncgen.py b/scrapy/utils/asyncgen.py
index c290e376c..9f794de92 100644
--- a/scrapy/utils/asyncgen.py
+++ b/scrapy/utils/asyncgen.py
@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterable
+from typing import AsyncGenerator, AsyncIterable, Iterable, Union
 
 
 async def collect_asyncgen(result: AsyncIterable):
@@ -6,3 +6,13 @@ async def collect_asyncgen(result: AsyncIterable):
     async for x in result:
         results.append(x)
     return results
+
+
+async def as_async_generator(it: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
+    """ Wraps an iterable (sync or async) into an async generator. """
+    if isinstance(it, AsyncIterable):
+        async for r in it:
+            yield r
+    else:
+        for r in it:
+            yield r
diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py
index 7ecb8ea3f..8fcf31cab 100644
--- a/scrapy/utils/defer.py
+++ b/scrapy/utils/defer.py
@@ -7,10 +7,15 @@ from asyncio import Future
 from functools import wraps
 from typing import (
     Any,
+    AsyncGenerator,
+    AsyncIterable,
     Callable,
     Coroutine,
     Generator,
     Iterable,
+    Iterator,
+    List,
+    Optional,
     Union
 )
 
@@ -87,6 +92,109 @@ def parallel(iterable: Iterable, count: int, callable: Callable, *args, **named)
     return DeferredList([coop.coiterate(work) for _ in range(count)])
 
 
+class _AsyncCooperatorAdapter(Iterator):
+    """ A class that wraps an async iterable into a normal iterator suitable
+    for using in Cooperator.coiterate(). As it's only needed for parallel_async(),
+    it calls the callable directly in the callback, instead of providing a more
+    generic interface.
+
+    On the outside, this class behaves as an iterator that yields Deferreds.
+    Each Deferred is fired with the result of the callable which was called on
+    the next result from aiterator. It raises StopIteration when aiterator is
+    exhausted, as expected.
+
+    Cooperator calls __next__() multiple times and waits on the Deferreds
+    returned from it. As async generators (since Python 3.8) don't support
+    awaiting on __anext__() several times in parallel, we need to serialize
+    this. It's done by storing the Deferreds returned from __next__() and
+    firing the oldest one when a result from __anext__() is available.
+
+    The workflow:
+    1. When __next__() is called for the first time, it creates a Deferred, stores it
+    in self.waiting_deferreds and returns it. It also makes a Deferred that will wait
+    for self.aiterator.__anext__() and puts it into self.anext_deferred.
+    2. If __next__() is called again before self.anext_deferred fires, more Deferreds
+    are added to self.waiting_deferreds.
+    3. When self.anext_deferred fires, it either calls _callback() or _errback(). Both
+    clear self.anext_deferred.
+    3.1. _callback() calls the callable passing the result value that it takes, pops a
+    Deferred from self.waiting_deferreds, and if the callable result was a Deferred, it
+    chains those Deferreds so that the waiting Deferred will fire when the result
+    Deferred does, otherwise it fires it directly. This causes one awaiting task to
+    receive a result. If self.waiting_deferreds is still not empty, new __anext__() is
+    called and self.anext_deferred is populated.
+    3.2. _errback() checks the exception class. If it's StopAsyncIteration it means
+    self.aiterator is exhausted and so it sets self.finished and fires all
+    self.waiting_deferreds. Other exceptions are propagated.
+    4. If __next__() is called after __anext__() was handled, then if self.finished is
+    True, it raises StopIteration, otherwise it acts like in step 2, but if
+    self.anext_deferred is now empty is also populates it with a new __anext__().
+
+    Note that CooperativeTask ignores the value returned from the Deferred that it waits
+    for, so we fire them with None when needed.
+
+    It may be possible to write an async iterator-aware replacement for
+    Cooperator/CooperativeTask and use it instead of this adapter to achieve the same
+    goal.
+    """
+    def __init__(self, aiterable: AsyncIterable, callable: Callable, *callable_args, **callable_kwargs):
+        self.aiterator = aiterable.__aiter__()
+        self.callable = callable
+        self.callable_args = callable_args
+        self.callable_kwargs = callable_kwargs
+        self.finished = False
+        self.waiting_deferreds: List[Deferred] = []
+        self.anext_deferred: Optional[Deferred] = None
+
+    def _callback(self, result: Any) -> None:
+        # This gets called when the result from aiterator.__anext__() is available.
+        # It calls the callable on it and sends the result to the oldest waiting Deferred
+        # (by chaining if the result is a Deferred too or by firing if not).
+        self.anext_deferred = None
+        result = self.callable(result, *self.callable_args, **self.callable_kwargs)
+        d = self.waiting_deferreds.pop(0)
+        if isinstance(result, Deferred):
+            result.chainDeferred(d)
+        else:
+            d.callback(None)
+        if self.waiting_deferreds:
+            self._call_anext()
+
+    def _errback(self, failure: Failure) -> None:
+        # This gets called on any exceptions in aiterator.__anext__().
+        # It handles StopAsyncIteration by stopping the iteration and reraises all others.
+        self.anext_deferred = None
+        failure.trap(StopAsyncIteration)
+        self.finished = True
+        for d in self.waiting_deferreds:
+            d.callback(None)
+
+    def _call_anext(self) -> None:
+        # This starts waiting for the next result from aiterator.
+        # If aiterator is exhausted, _errback will be called.
+        self.anext_deferred = deferred_from_coro(self.aiterator.__anext__())
+        self.anext_deferred.addCallbacks(self._callback, self._errback)
+
+    def __next__(self) -> Deferred:
+        # This puts a new Deferred into self.waiting_deferreds and returns it.
+        # It also calls __anext__() if needed.
+        if self.finished:
+            raise StopIteration
+        d = Deferred()
+        self.waiting_deferreds.append(d)
+        if not self.anext_deferred:
+            self._call_anext()
+        return d
+
+
+def parallel_async(async_iterable: AsyncIterable, count: int, callable: Callable, *args, **named) -> DeferredList:
+    """ Like parallel but for async iterators """
+    coop = Cooperator()
+    work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named)
+    dl = DeferredList([coop.coiterate(work) for _ in range(count)])
+    return dl
+
+
 def process_chain(callbacks: Iterable[Callable], input, *a, **kw) -> Deferred:
     """Return a Deferred built by chaining the given callbacks"""
     d = Deferred()
@@ -136,6 +244,20 @@ def iter_errback(iterable: Iterable, errback: Callable, *a, **kw) -> Generator:
             errback(failure.Failure(), *a, **kw)
 
 
+async def aiter_errback(aiterable: AsyncIterable, errback: Callable, *a, **kw) -> AsyncGenerator:
+    """Wraps an async iterable calling an errback if an error is caught while
+    iterating it. Similar to scrapy.utils.defer.iter_errback()
+    """
+    it = aiterable.__aiter__()
+    while True:
+        try:
+            yield await it.__anext__()
+        except StopAsyncIteration:
+            break
+        except Exception:
+            errback(failure.Failure(), *a, **kw)
+
+
 def deferred_from_coro(o) -> Any:
     """Converts a coroutine into a Deferred, or returns the object as is if it isn't a coroutine"""
     if isinstance(o, Deferred):
diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py
index bcc12f24f..11c089ac2 100644
--- a/scrapy/utils/python.py
+++ b/scrapy/utils/python.py
@@ -8,11 +8,12 @@ import re
 import sys
 import warnings
 import weakref
-from collections.abc import Iterable
 from functools import partial, wraps
 from itertools import chain
+from typing import AsyncGenerator, AsyncIterable, Iterable, Union
 
 from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.utils.asyncgen import as_async_generator
 from scrapy.utils.decorators import deprecated
 
 
@@ -344,7 +345,7 @@ class MutableChain(Iterable):
     def __init__(self, *args: Iterable):
         self.data = chain.from_iterable(args)
 
-    def extend(self, *iterables: Iterable):
+    def extend(self, *iterables: Iterable) -> None:
         self.data = chain(self.data, chain.from_iterable(iterables))
 
     def __iter__(self):
@@ -356,3 +357,27 @@ class MutableChain(Iterable):
     @deprecated("scrapy.utils.python.MutableChain.__next__")
     def next(self):
         return self.__next__()
+
+
+async def _async_chain(*iterables: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
+    for it in iterables:
+        async for o in as_async_generator(it):
+            yield o
+
+
+class MutableAsyncChain(AsyncIterable):
+    """
+    Similar to MutableChain but for async iterables
+    """
+
+    def __init__(self, *args: Union[Iterable, AsyncIterable]):
+        self.data = _async_chain(*args)
+
+    def extend(self, *iterables: Union[Iterable, AsyncIterable]) -> None:
+        self.data = _async_chain(self.data, _async_chain(*iterables))
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        return await self.data.__anext__()
diff --git a/scrapy/utils/spider.py b/scrapy/utils/spider.py
index 59fc9202f..d0fd1757d 100644
--- a/scrapy/utils/spider.py
+++ b/scrapy/utils/spider.py
@@ -4,7 +4,6 @@ import logging
 from scrapy.spiders import Spider
 from scrapy.utils.defer import deferred_from_coro
 from scrapy.utils.misc import arg_to_iter
-from scrapy.utils.asyncgen import collect_asyncgen
 
 
 logger = logging.getLogger(__name__)
@@ -12,14 +11,13 @@ logger = logging.getLogger(__name__)
 
 def iterate_spider_output(result):
     if inspect.isasyncgen(result):
-        d = deferred_from_coro(collect_asyncgen(result))
-        d.addCallback(iterate_spider_output)
-        return d
+        return result
     elif inspect.iscoroutine(result):
         d = deferred_from_coro(result)
         d.addCallback(iterate_spider_output)
         return d
-    return arg_to_iter(result)
+    else:
+        return arg_to_iter(deferred_from_coro(result))
 
 
 def iter_spider_classes(module):
diff --git a/tox.ini b/tox.ini
index 4d1bb574d..2110e1020 100644
--- a/tox.ini
+++ b/tox.ini
@@ -38,7 +38,7 @@ install_command =
 basepython = python3
 deps =
     lxml-stubs==0.2.0
-    mypy==0.910
+    mypy==0.971
     types-pyOpenSSL==20.0.3
     types-setuptools==57.0.0
 commands =


Test patch:
diff --git a/scrapy/utils/test.py b/scrapy/utils/test.py
index b90ea5009..0b828f7c0 100644
--- a/scrapy/utils/test.py
+++ b/scrapy/utils/test.py
@@ -115,3 +115,10 @@ def mock_google_cloud_storage():
     bucket_mock.blob.return_value = blob_mock
 
     return (client_mock, bucket_mock, blob_mock)
+
+
+def get_web_client_agent_req(url):
+    from twisted.internet import reactor
+    from twisted.web.client import Agent  # imports twisted.internet.reactor
+    agent = Agent(reactor)
+    return agent.request(b'GET', url.encode('utf-8'))
diff --git a/tests/spiders.py b/tests/spiders.py
index 67dbbbe0f..3b69aa7ae 100644
--- a/tests/spiders.py
+++ b/tests/spiders.py
@@ -14,7 +14,8 @@ from scrapy.item import Item
 from scrapy.linkextractors import LinkExtractor
 from scrapy.spiders import Spider
 from scrapy.spiders.crawl import CrawlSpider, Rule
-from scrapy.utils.test import get_from_asyncio_queue
+from scrapy.utils.defer import deferred_to_future, maybe_deferred_to_future
+from scrapy.utils.test import get_from_asyncio_queue, get_web_client_agent_req
 
 
 class MockServerSpider(Spider):
@@ -148,6 +149,41 @@ class AsyncDefAsyncioReqsReturnSpider(SimpleSpider):
         return reqs
 
 
+class AsyncDefAsyncioGenExcSpider(SimpleSpider):
+    name = 'asyncdef_asyncio_gen_exc'
+
+    async def parse(self, response):
+        for i in range(10):
+            await asyncio.sleep(0.1)
+            yield {'foo': i}
+            if i > 5:
+                raise ValueError("Stopping the processing")
+
+
+class AsyncDefDeferredDirectSpider(SimpleSpider):
+    name = 'asyncdef_deferred_direct'
+
+    async def parse(self, response):
+        resp = await get_web_client_agent_req(self.mockserver.url("/status?n=200"))
+        yield {'code': resp.code}
+
+
+class AsyncDefDeferredWrappedSpider(SimpleSpider):
+    name = 'asyncdef_deferred_wrapped'
+
+    async def parse(self, response):
+        resp = await deferred_to_future(get_web_client_agent_req(self.mockserver.url("/status?n=200")))
+        yield {'code': resp.code}
+
+
+class AsyncDefDeferredMaybeWrappedSpider(SimpleSpider):
+    name = 'asyncdef_deferred_wrapped'
+
+    async def parse(self, response):
+        resp = await maybe_deferred_to_future(get_web_client_agent_req(self.mockserver.url("/status?n=200")))
+        yield {'code': resp.code}
+
+
 class AsyncDefAsyncioGenSpider(SimpleSpider):
 
     name = 'asyncdef_asyncio_gen'
diff --git a/tests/test_crawl.py b/tests/test_crawl.py
index 59c271868..c11871745 100644
--- a/tests/test_crawl.py
+++ b/tests/test_crawl.py
@@ -23,12 +23,16 @@ from tests import NON_EXISTING_RESOLVABLE
 from tests.mockserver import MockServer
 from tests.spiders import (
     AsyncDefAsyncioGenComplexSpider,
+    AsyncDefAsyncioGenExcSpider,
     AsyncDefAsyncioGenLoopSpider,
     AsyncDefAsyncioGenSpider,
     AsyncDefAsyncioReqsReturnSpider,
     AsyncDefAsyncioReturnSingleElementSpider,
     AsyncDefAsyncioReturnSpider,
     AsyncDefAsyncioSpider,
+    AsyncDefDeferredDirectSpider,
+    AsyncDefDeferredMaybeWrappedSpider,
+    AsyncDefDeferredWrappedSpider,
     AsyncDefSpider,
     BrokenStartRequestsSpider,
     BytesReceivedCallbackSpider,
@@ -458,6 +462,18 @@ class CrawlSpiderTestCase(TestCase):
         for i in range(10):
             self.assertIn({'foo': i}, items)
 
+    @mark.only_asyncio()
+    @defer.inlineCallbacks
+    def test_async_def_asyncgen_parse_exc(self):
+        log, items, stats = yield self._run_spider(AsyncDefAsyncioGenExcSpider)
+        log = str(log)
+        self.assertIn("Spider error processing", log)
+        self.assertIn("ValueError", log)
+        itemcount = stats.get_value('item_scraped_count')
+        self.assertEqual(itemcount, 7)
+        for i in range(7):
+            self.assertIn({'foo': i}, items)
+
     @mark.only_asyncio()
     @defer.inlineCallbacks
     def test_async_def_asyncgen_parse_complex(self):
@@ -477,6 +493,23 @@ class CrawlSpiderTestCase(TestCase):
         for req_id in range(3):
             self.assertIn(f"Got response 200, req_id {req_id}", str(log))
 
+    @mark.only_not_asyncio()
+    @defer.inlineCallbacks
+    def test_async_def_deferred_direct(self):
+        _, items, _ = yield self._run_spider(AsyncDefDeferredDirectSpider)
+        self.assertEqual(items, [{'code': 200}])
+
+    @mark.only_asyncio()
+    @defer.inlineCallbacks
+    def test_async_def_deferred_wrapped(self):
+        log, items, _ = yield self._run_spider(AsyncDefDeferredWrappedSpider)
+        self.assertEqual(items, [{'code': 200}])
+
+    @defer.inlineCallbacks
+    def test_async_def_deferred_maybe_wrapped(self):
+        _, items, _ = yield self._run_spider(AsyncDefDeferredMaybeWrappedSpider)
+        self.assertEqual(items, [{'code': 200}])
+
     @defer.inlineCallbacks
     def test_response_ssl_certificate_none(self):
         crawler = get_crawler(SingleRequestSpider)
diff --git a/tests/test_spidermiddleware.py b/tests/test_spidermiddleware.py
index b39576996..edde6f682 100644
--- a/tests/test_spidermiddleware.py
+++ b/tests/test_spidermiddleware.py
@@ -1,11 +1,17 @@
+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
 
 from scrapy.spiders import Spider
 from scrapy.http import Request, Response
 from scrapy.exceptions import _InvalidOutput
+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
 
@@ -101,3 +107,427 @@ class ProcessSpiderExceptionReRaise(SpiderMiddlewareTestCase):
         result = self._scrape_response()
         self.assertIsInstance(result, Failure)
         self.assertIsInstance(result.value, ZeroDivisionError)
+
+
+class BaseAsyncSpiderMiddlewareTestCase(SpiderMiddlewareTestCase):
+    """ Helpers for testing sync, async and mixed middlewares.
+
+    Should work for process_spider_output and, when it's supported, process_start_requests.
+    """
+
+    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)}
+
+    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_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, 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, 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)
+        self.assertEqual("downgraded to a non-async" in str(log), downgrade)
+
+
+class ProcessSpiderOutputSimpleMiddleware:
+    def process_spider_output(self, response, result, spider):
+        for r in result:
+            yield r
+
+
+class ProcessSpiderOutputAsyncGenMiddleware:
+    async def process_spider_output(self, response, result, spider):
+        async for r in result:
+            yield r
+
+
+class ProcessSpiderOutputUniversalMiddleware:
+    def process_spider_output(self, response, result, spider):
+        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):
+    """ process_spider_output tests for simple callbacks"""
+
+    ITEM_TYPE = dict
+    MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
+    MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
+    MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
+
+    def test_simple(self):
+        """ Simple mw """
+        return self._test_simple_base(self.MW_SIMPLE)
+
+    def test_asyncgen(self):
+        """ Asyncgen mw; upgrade """
+        return self._test_asyncgen_base(self.MW_ASYNCGEN)
+
+    def test_simple_asyncgen(self):
+        """ 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; upgrade then downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_ASYNCGEN,
+                                      downgrade=True)
+
+    def test_universal(self):
+        """ Universal mw """
+        return self._test_simple_base(self.MW_UNIVERSAL)
+
+    def test_universal_simple(self):
+        """ Universal mw -> simple mw """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_UNIVERSAL)
+
+    def test_simple_universal(self):
+        """ Simple mw -> universal mw """
+        return self._test_simple_base(self.MW_UNIVERSAL,
+                                      self.MW_SIMPLE)
+
+    def test_universal_asyncgen(self):
+        """ 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; upgrade """
+        return self._test_asyncgen_base(self.MW_UNIVERSAL,
+                                        self.MW_ASYNCGEN)
+
+
+class ProcessSpiderOutputAsyncGen(ProcessSpiderOutputSimple):
+    """ process_spider_output tests for async generator callbacks """
+
+    async def _scrape_func(self, *args, **kwargs):
+        for item in super()._scrape_func():
+            yield item
+
+    def test_simple(self):
+        """ Simple mw; downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      downgrade=True)
+
+    def test_simple_asyncgen(self):
+        """ 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; downgrade """
+        return self._test_simple_base(self.MW_SIMPLE,
+                                      self.MW_UNIVERSAL,
+                                      downgrade=True)
+
+    def test_simple_universal(self):
+        """ Simple mw -> universal mw; downgrade """
+        return self._test_simple_base(self.MW_UNIVERSAL,
+                                      self.MW_SIMPLE,
+                                      downgrade=True)
+
+
+class ProcessSpiderOutputNonIterableMiddleware:
+    def process_spider_output(self, response, result, spider):
+        return
+
+
+class ProcessSpiderOutputCoroutineMiddleware:
+    async def process_spider_output(self, response, result, spider):
+        results = []
+        for r in result:
+            results.append(r)
+        return results
+
+
+class ProcessSpiderOutputInvalidResult(BaseAsyncSpiderMiddlewareTestCase):
+
+    @defer.inlineCallbacks
+    def test_non_iterable(self):
+        with self.assertRaisesRegex(
+            _InvalidOutput,
+            (
+                r"\.process_spider_output must return an iterable, got <class "
+                r"'NoneType'>"
+            ),
+        ):
+            yield self._get_middleware_result(
+                ProcessSpiderOutputNonIterableMiddleware,
+            )
+
+    @defer.inlineCallbacks
+    def test_coroutine(self):
+        with self.assertRaisesRegex(
+            _InvalidOutput,
+            r"\.process_spider_output must be an asynchronous generator",
+        ):
+            yield self._get_middleware_result(
+                ProcessSpiderOutputCoroutineMiddleware,
+            )
+
+
+class ProcessStartRequestsSimpleMiddleware:
+    def process_start_requests(self, start_requests, spider):
+        for r in start_requests:
+            yield r
+
+
+class ProcessStartRequestsSimple(BaseAsyncSpiderMiddlewareTestCase):
+    """ process_start_requests tests for simple start_requests"""
+
+    ITEM_TYPE = Request
+    MW_SIMPLE = ProcessStartRequestsSimpleMiddleware
+
+    def _start_requests(self):
+        for i in range(3):
+            yield Request(f'https://example.com/{i}', dont_filter=True)
+
+    @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_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 """
+        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
+
+    @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)
diff --git a/tests/test_spidermiddleware_output_chain.py b/tests/test_spidermiddleware_output_chain.py
index 029bf8bd6..dac246fb6 100644
--- a/tests/test_spidermiddleware_output_chain.py
+++ b/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,
         },
@@ -43,6 +44,14 @@ class RecoverySpider(Spider):
             raise TabError()
 
 
+class RecoveryAsyncGenSpider(RecoverySpider):
+    name = 'RecoveryAsyncGenSpider'
+
+    async def parse(self, response):
+        for r in super().parse(response):
+            yield r
+
+
 # ================================================================================
 # (1) exceptions from a spider middleware's process_spider_input method
 class FailProcessSpiderInputMiddleware:
@@ -99,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):
@@ -307,6 +323,16 @@ class TestSpiderMiddleware(TestCase):
         self.assertEqual(str(log).count("Middleware: TabError exception caught"), 1)
         self.assertIn("'item_scraped_count': 3", str(log))
 
+    @defer.inlineCallbacks
+    def test_recovery_asyncgen(self):
+        """
+        Same as test_recovery but with an async callback.
+        """
+        log = yield self.crawl_log(RecoveryAsyncGenSpider)
+        self.assertIn("Middleware: TabError exception caught", str(log))
+        self.assertEqual(str(log).count("Middleware: TabError exception caught"), 1)
+        self.assertIn("'item_scraped_count': 3", str(log))
+
     @defer.inlineCallbacks
     def test_process_spider_input_without_errback(self):
         """
@@ -342,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):
         """
diff --git a/tests/test_utils_asyncgen.py b/tests/test_utils_asyncgen.py
new file mode 100644
index 000000000..9ae66c57c
--- /dev/null
+++ b/tests/test_utils_asyncgen.py
@@ -0,0 +1,20 @@
+from twisted.trial import unittest
+
+from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
+from scrapy.utils.defer import deferred_f_from_coro_f
+
+
+class AsyncgenUtilsTest(unittest.TestCase):
+    @deferred_f_from_coro_f
+    async def test_as_async_generator(self):
+        ag = as_async_generator(range(42))
+        results = []
+        async for i in ag:
+            results.append(i)
+        self.assertEqual(results, list(range(42)))
+
+    @deferred_f_from_coro_f
+    async def test_collect_asyncgen(self):
+        ag = as_async_generator(range(42))
+        results = await collect_asyncgen(ag)
+        self.assertEqual(results, list(range(42)))
diff --git a/tests/test_utils_defer.py b/tests/test_utils_defer.py
index 032dbc8c5..d39de7430 100644
--- a/tests/test_utils_defer.py
+++ b/tests/test_utils_defer.py
@@ -1,12 +1,18 @@
+import random
+
 from pytest import mark
 from twisted.trial import unittest
 from twisted.internet import reactor, defer
 from twisted.python.failure import Failure
 
+from scrapy.utils.asyncgen import collect_asyncgen, as_async_generator
 from scrapy.utils.defer import (
+    aiter_errback,
     deferred_f_from_coro_f,
     iter_errback,
+    maybe_deferred_to_future,
     mustbe_deferred,
+    parallel_async,
     process_chain,
     process_chain_both,
     process_parallel,
@@ -121,6 +127,34 @@ class IterErrbackTest(unittest.TestCase):
         self.assertIsInstance(errors[0].value, ZeroDivisionError)
 
 
+class AiterErrbackTest(unittest.TestCase):
+
+    @deferred_f_from_coro_f
+    async def test_aiter_errback_good(self):
+        async def itergood():
+            for x in range(10):
+                yield x
+
+        errors = []
+        out = await collect_asyncgen(aiter_errback(itergood(), errors.append))
+        self.assertEqual(out, list(range(10)))
+        self.assertFalse(errors)
+
+    @deferred_f_from_coro_f
+    async def test_iter_errback_bad(self):
+        async def iterbad():
+            for x in range(10):
+                if x == 5:
+                    1 / 0
+                yield x
+
+        errors = []
+        out = await collect_asyncgen(aiter_errback(iterbad(), errors.append))
+        self.assertEqual(out, [0, 1, 2, 3, 4])
+        self.assertEqual(len(errors), 1)
+        self.assertIsInstance(errors[0].value, ZeroDivisionError)
+
+
 class AsyncDefTestsuiteTest(unittest.TestCase):
     @deferred_f_from_coro_f
     async def test_deferred_f_from_coro_f(self):
@@ -134,3 +168,65 @@ class AsyncDefTestsuiteTest(unittest.TestCase):
     @deferred_f_from_coro_f
     async def test_deferred_f_from_coro_f_xfail(self):
         raise Exception("This is expected to be raised")
+
+
+class AsyncCooperatorTest(unittest.TestCase):
+    """ This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
+
+    parallel_async is called with the results of a callback (so an iterable of items, requests and None,
+    with arbitrary delays between values), and it uses Scraper._process_spidermw_output as the callable
+    (so a callable that returns a Deferred for an item, which will fire after pipelines process it, and
+    None for everything else). The concurrent task count is the CONCURRENT_ITEMS setting.
+
+    We want to test different concurrency values compared to the iterable length.
+    We also want to simulate the real usage, with arbitrary delays between getting the values
+    from the iterable. We also want to simulate sync and async results from the callable.
+    """
+    CONCURRENT_ITEMS = 50
+
+    @staticmethod
+    def callable(o, results):
+        if random.random() < 0.4:
+            # simulate async processing
+            dfd = defer.Deferred()
+            dfd.addCallback(lambda _: results.append(o))
+            delay = random.random() / 8
+            reactor.callLater(delay, dfd.callback, None)
+            return dfd
+        else:
+            # simulate trivial sync processing
+            results.append(o)
+
+    @staticmethod
+    def get_async_iterable(length):
+        # simulate a simple callback without delays between results
+        return as_async_generator(range(length))
+
+    @staticmethod
+    async def get_async_iterable_with_delays(length):
+        # simulate a callback with delays between some of the results
+        for i in range(length):
+            if random.random() < 0.1:
+                dfd = defer.Deferred()
+                delay = random.random() / 20
+                reactor.callLater(delay, dfd.callback, None)
+                await maybe_deferred_to_future(dfd)
+            yield i
+
+    @defer.inlineCallbacks
+    def test_simple(self):
+        for length in [20, 50, 100]:
+            results = []
+            ait = self.get_async_iterable(length)
+            dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
+            yield dl
+            self.assertEqual(list(range(length)), sorted(results))
+
+    @defer.inlineCallbacks
+    def test_delays(self):
+        for length in [20, 50, 100]:
+            results = []
+            ait = self.get_async_iterable_with_delays(length)
+            dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
+            yield dl
+            self.assertEqual(list(range(length)), sorted(results))
diff --git a/tests/test_utils_python.py b/tests/test_utils_python.py
index 7dec5624a..b1a8fdc04 100644
--- a/tests/test_utils_python.py
+++ b/tests/test_utils_python.py
@@ -2,15 +2,18 @@ import functools
 import gc
 import operator
 import platform
-import unittest
 from itertools import count
 from warnings import catch_warnings, filterwarnings
 
+from twisted.trial import unittest
+
 from scrapy.exceptions import ScrapyDeprecationWarning
+from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
+from scrapy.utils.defer import deferred_f_from_coro_f, aiter_errback
 from scrapy.utils.python import (
     memoizemethod_noargs, binary_is_text, equal_attributes,
     WeakKeyCache, get_func_args, to_bytes, to_unicode,
-    without_none_values, MutableChain)
+    without_none_values, MutableChain, MutableAsyncChain)
 
 
 __doctests__ = ['scrapy.utils.python']
@@ -32,6 +35,57 @@ class MutableChainTest(unittest.TestCase):
         self.assertEqual(list(m), list(range(3, 13)))
 
 
+class MutableAsyncChainTest(unittest.TestCase):
+    @staticmethod
+    async def g1():
+        for i in range(3):
+            yield i
+
+    @staticmethod
+    async def g2():
+        return
+        yield
+
+    @staticmethod
+    async def g3():
+        for i in range(7, 10):
+            yield i
+
+    @staticmethod
+    async def g4():
+        for i in range(3, 5):
+            yield i
+        1 / 0
+        for i in range(5, 7):
+            yield i
+
+    @staticmethod
+    async def collect_asyncgen_exc(asyncgen):
+        results = []
+        async for x in asyncgen:
+            results.append(x)
+        return results
+
+    @deferred_f_from_coro_f
+    async def test_mutableasyncchain(self):
+        m = MutableAsyncChain(self.g1(), as_async_generator(range(3, 7)))
+        m.extend(self.g2())
+        m.extend(self.g3())
+
+        self.assertEqual(await m.__anext__(), 0)
+        results = await collect_asyncgen(m)
+        self.assertEqual(results, list(range(1, 10)))
+
+    @deferred_f_from_coro_f
+    async def test_mutableasyncchain_exc(self):
+        m = MutableAsyncChain(self.g1())
+        m.extend(self.g4())
+        m.extend(self.g3())
+
+        results = await collect_asyncgen(aiter_errback(m, lambda _: None))
+        self.assertEqual(results, list(range(5)))
+
+
 class ToUnicodeTest(unittest.TestCase):
     def test_converting_an_utf8_encoded_string_to_unicode(self):
         self.assertEqual(to_unicode(b'lel\xc3\xb1e'), 'lel\xf1e')
diff --git a/tests/test_utils_request.py b/tests/test_utils_request.py
index 5ee772c0b..8bc7922b6 100644
--- a/tests/test_utils_request.py
+++ b/tests/test_utils_request.py
@@ -52,7 +52,7 @@ class UtilsRequestTest(unittest.TestCase):
 class FingerprintTest(unittest.TestCase):
     maxDiff = None
 
-    function = staticmethod(fingerprint)
+    function: staticmethod = staticmethod(fingerprint)
     cache: Union[
         "WeakKeyDictionary[Request, Dict[Tuple[Optional[Tuple[bytes, ...]], bool], bytes]]",
         "WeakKeyDictionary[Request, Dict[Tuple[Optional[Tuple[bytes, ...]], bool], str]]",