diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index 0d3e3450f..3eae71af7 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -1,6 +1,6 @@ """This module implements the Scraper component which parses responses and extracts information from them""" - +import collections import logging from collections import deque @@ -12,7 +12,16 @@ from scrapy import signals 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, + deferred_from_coro, + 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 @@ -180,9 +189,14 @@ class Scraper: def handle_spider_output(self, result, request, response, spider): 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) + if isinstance(result, collections.abc.AsyncIterable): + it = aiter_errback(result, self.handle_spider_error, request, response, spider) + dfd = deferred_from_coro(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, request, response, spider): diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index 2d02c0621..bd5f9c8fc 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -75,6 +75,116 @@ def parallel(iterable, count, callable, *args, **named): return defer.DeferredList([coop.coiterate(work) for _ in range(count)]) +class _AsyncCooperatorAdapter: + """ A class that wraps an async iterator 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, aiterator, callable, *callable_args, **callable_kwargs): + self.aiterator = aiterator + self.callable = callable + self.callable_args = callable_args + self.callable_kwargs = callable_kwargs + self.finished = False + self.waiting_deferreds = [] + self.anext_deferred = None + + def _callback(self, result): + # 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 d.called: + raise ValueError('Deferred in waiting_deferreds already called') + if isinstance(result, defer.Deferred): + result.chainDeferred(d) + else: + d.callback(None) + if self.waiting_deferreds: + self._call_anext() + + def _errback(self, failure): + # 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: + if d.called: + raise ValueError('Deferred in waiting_deferreds already called') + d.callback(None) + + def _call_anext(self): + # 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 __iter__(self): + return self + + def __next__(self): + # This puts a new Deferred into self.waiting_deferreds and returns it. + # It also calls __anext__() if needed. + if self.finished: + raise StopIteration + d = defer.Deferred() + self.waiting_deferreds.append(d) + if not self.anext_deferred: + self._call_anext() + return d + + +def parallel_async(async_iterable, count, callable, *args, **named): + """ Like parallel but for async iterables """ + coop = task.Cooperator() + work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named) + dl = defer.DeferredList([coop.coiterate(work) for _ in range(count)]) + return dl + + def process_chain(callbacks, input, *a, **kw): """Return a Deferred built by chaining the given callbacks""" d = defer.Deferred()