diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py index d3d7e5f8c..33e215971 100644 --- a/scrapy/core/spidermw.py +++ b/scrapy/core/spidermw.py @@ -10,6 +10,7 @@ from twisted.python.failure import Failure from scrapy.exceptions import _InvalidOutput from scrapy.middleware import MiddlewareManager +from scrapy.utils.asyncgen import _process_iterable_universal from scrapy.utils.conf import build_component_list from scrapy.utils.defer import mustbe_deferred from scrapy.utils.python import MutableAsyncChain, MutableChain @@ -57,31 +58,18 @@ class SpiderMiddlewareManager(MiddlewareManager): return scrape_func(response, request, spider) def _evaluate_iterable(self, response, spider, iterable, exception_processor_index, recover_to): - def _process_exception(ex): - exception_result = self._process_spider_exception(response, spider, Failure(ex), - exception_processor_index) - if isinstance(exception_result, Failure): - raise # pylint: disable=E0704 - recover_to.extend(exception_result) - - def _evaluate_normal_iterable(iterable): - try: - for r in iterable: - yield r - except Exception as ex: - _process_exception(ex) - + @_process_iterable_universal async def _evaluate_async_iterable(iterable): try: async for r in iterable: yield r except Exception as ex: - _process_exception(ex) - - if inspect.isasyncgen(iterable): - return _evaluate_async_iterable(iterable) - else: - return _evaluate_normal_iterable(iterable) + exception_result = self._process_spider_exception(response, spider, Failure(ex), + exception_processor_index) + if isinstance(exception_result, Failure): + raise + recover_to.extend(exception_result) + return _evaluate_async_iterable(iterable) def _process_spider_exception(self, response, spider, _failure, start_index=0): exception = _failure.value diff --git a/scrapy/spidermiddlewares/depth.py b/scrapy/spidermiddlewares/depth.py index 776a6879a..973404b2b 100644 --- a/scrapy/spidermiddlewares/depth.py +++ b/scrapy/spidermiddlewares/depth.py @@ -7,6 +7,7 @@ See documentation in docs/topics/spider-middleware.rst import logging from scrapy.http import Request +from scrapy.utils.asyncgen import _process_iterable_universal logger = logging.getLogger(__name__) @@ -49,10 +50,15 @@ class DepthMiddleware: spider=spider) return True - # 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) + @_process_iterable_universal + async def process(result): + # base case (depth=0) + if 'depth' not in response.meta: + response.meta['depth'] = 0 + if self.verbose_stats: + self.stats.inc_value('request_depth_count/0', spider=spider) - return (r for r in result or () if _filter(r)) + async for r in result or (): + if _filter(r): + yield r + return process(result) diff --git a/scrapy/spidermiddlewares/offsite.py b/scrapy/spidermiddlewares/offsite.py index 6e4efda97..074ec7a4e 100644 --- a/scrapy/spidermiddlewares/offsite.py +++ b/scrapy/spidermiddlewares/offsite.py @@ -9,6 +9,7 @@ import warnings from scrapy import signals from scrapy.http import Request +from scrapy.utils.asyncgen import _process_iterable_universal from scrapy.utils.httpobj import urlparse_cached logger = logging.getLogger(__name__) @@ -26,21 +27,24 @@ 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 + @_process_iterable_universal + async def process(result): + async for x in result: + if isinstance(x, Request): + if x.dont_filter or self.should_follow(x, spider): + yield x + else: + domain = urlparse_cached(x).hostname + if domain and domain not in self.domains_seen: + self.domains_seen.add(domain) + logger.debug( + "Filtered offsite request to %(domain)r: %(request)s", + {'domain': domain, 'request': x}, extra={'spider': spider}) + self.stats.inc_value('offsite/domains', spider=spider) + self.stats.inc_value('offsite/filtered', spider=spider) else: - 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 + yield x + return process(result) def should_follow(self, request, spider): regex = self.host_regex diff --git a/scrapy/spidermiddlewares/referer.py b/scrapy/spidermiddlewares/referer.py index f81041376..8d862d1d0 100644 --- a/scrapy/spidermiddlewares/referer.py +++ b/scrapy/spidermiddlewares/referer.py @@ -10,6 +10,7 @@ from w3lib.url import safe_url_string from scrapy.http import Request, Response from scrapy.exceptions import NotConfigured from scrapy import signals +from scrapy.utils.asyncgen import _process_iterable_universal from scrapy.utils.python import to_unicode from scrapy.utils.misc import load_object from scrapy.utils.url import strip_url @@ -337,7 +338,13 @@ class RefererMiddleware: if referrer is not None: r.headers.setdefault('Referer', referrer) return r - return (_set_referer(r) for r in result or ()) + + @_process_iterable_universal + async def process(result): + async for r in result or (): + yield _set_referer(r) + + return process(result) 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 5be1f80cb..d40d43ff1 100644 --- a/scrapy/spidermiddlewares/urllength.py +++ b/scrapy/spidermiddlewares/urllength.py @@ -8,6 +8,7 @@ import logging from scrapy.http import Request from scrapy.exceptions import NotConfigured +from scrapy.utils.asyncgen import _process_iterable_universal logger = logging.getLogger(__name__) @@ -34,4 +35,10 @@ class UrlLengthMiddleware: else: return True - return (r for r in result or () if _filter(r)) + @_process_iterable_universal + async def process(result): + async for r in result or (): + if _filter(r): + yield r + + return process(result) diff --git a/scrapy/utils/asyncgen.py b/scrapy/utils/asyncgen.py index db2173f85..39c94ad8a 100644 --- a/scrapy/utils/asyncgen.py +++ b/scrapy/utils/asyncgen.py @@ -1,4 +1,6 @@ import collections +import functools +import inspect async def collect_asyncgen(result): @@ -15,3 +17,46 @@ async def as_async_generator(it): else: for r in it: yield r + + +# https://stackoverflow.com/a/66170760/113586 +def _process_iterable_universal(process_async): + """ Takes a function that takes an async iterable, args and kwargs. Returns + a function that takes any iterable, args and kwargs. + + Requires that process_async only awaits on the iterable and synchronous functions, + so it's better to use this only in the Scrapy code itself. + """ + + # If this stops working, all internal uses can be just replaced with manually-written + # process_sync functions. + + def process_sync(iterable, *args, **kwargs): + agen = process_async(as_async_generator(iterable), *args, **kwargs) + if not inspect.isasyncgen(agen): + raise ValueError(f"process_async returned wrong type {type(agen)}") + sent = None + while True: + try: + gen = agen.asend(sent) + gen.send(None) + except StopIteration as e: + sent = yield e.value + except StopAsyncIteration: + return + else: + gen.throw(RuntimeError, + f"Synchronously-called function '{process_async.__name__}' has blocked, " + f"you can't use {_process_iterable_universal.__name__} with it.") + + @functools.wraps(process_async) + def process(iterable, *args, **kwargs): + if inspect.isasyncgen(iterable): + # call process_async directly + return process_async(iterable, *args, **kwargs) + if hasattr(iterable, '__iter__'): + # convert process_async to process_sync + return process_sync(iterable, *args, **kwargs) + raise ValueError(f"Wrong iterable type {type(iterable)}") + + return process diff --git a/scrapy/utils/middlewares.py b/scrapy/utils/middlewares.py deleted file mode 100644 index da28e0ddf..000000000 --- a/scrapy/utils/middlewares.py +++ /dev/null @@ -1,35 +0,0 @@ -# coding: utf-8 -import inspect - - -def process_normal_iterable_helper(it, in_predicate=None, out_predicate=None, processor=None): - for o in it: - if in_predicate and not in_predicate(o): - continue - if processor is not None: - o = processor(o) - if out_predicate and not out_predicate(o): - continue - yield o - - -async def process_async_iterable_helper(it, in_predicate=None, out_predicate=None, processor=None): - async for o in it: - if in_predicate and not in_predicate(o): - continue - if processor is not None: - o = processor(o) - if out_predicate and not out_predicate(o): - continue - yield o - - -def process_iterable_helper(it, in_predicate=None, out_predicate=None, processor=None): - """ - For each item in the iterable: skips it if in_predicate is False, applies processor, - skips the result if out_predicate is False, else yields it. - """ - if inspect.isasyncgen(it): - return process_async_iterable_helper(it, in_predicate, out_predicate, processor) - else: - return process_normal_iterable_helper(it, in_predicate, out_predicate, processor) diff --git a/tests/test_utils_asyncgen.py b/tests/test_utils_asyncgen.py index 9ae66c57c..2f4181d3d 100644 --- a/tests/test_utils_asyncgen.py +++ b/tests/test_utils_asyncgen.py @@ -1,6 +1,6 @@ from twisted.trial import unittest -from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen +from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen, _process_iterable_universal from scrapy.utils.defer import deferred_f_from_coro_f @@ -18,3 +18,23 @@ class AsyncgenUtilsTest(unittest.TestCase): ag = as_async_generator(range(42)) results = await collect_asyncgen(ag) self.assertEqual(results, list(range(42))) + + +@_process_iterable_universal +async def process_iterable(iterable): + async for i in iterable: + yield i * 2 + + +class ProcessIterableUniversalTest(unittest.TestCase): + + def test_normal(self): + iterable = iter([1, 2, 3]) + results = list(process_iterable(iterable)) + self.assertEqual(results, [2, 4, 6]) + + @deferred_f_from_coro_f + async def test_async(self): + iterable = as_async_generator([1, 2, 3]) + results = await collect_asyncgen(process_iterable(iterable)) + self.assertEqual(results, [2, 4, 6]) diff --git a/tests/test_utils_middlewares.py b/tests/test_utils_middlewares.py deleted file mode 100644 index d395ba1a9..000000000 --- a/tests/test_utils_middlewares.py +++ /dev/null @@ -1,87 +0,0 @@ -import collections - -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 -from scrapy.utils.middlewares import process_iterable_helper - - -def predicate1(o): - return bool(o % 2) - - -def predicate2(o): - return o < 10 - - -def processor(o): - return o * 2 - - -class ProcessIterableHelperNormalTest(unittest.TestCase): - - def test_normal_in_predicate(self): - iterable1 = iter([1, 2, 3]) - iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1) - self.assertIsInstance(iterable2, collections.abc.Iterable) - list2 = list(iterable2) - self.assertEqual(list2, [1, 3]) - - def test_normal_out_predicate(self): - iterable1 = iter([1, 2, 10, 3, 15]) - iterable2 = process_iterable_helper(iterable1, out_predicate=predicate2) - self.assertIsInstance(iterable2, collections.abc.Iterable) - list2 = list(iterable2) - self.assertEqual(list2, [1, 2, 3]) - - def test_normal_processor(self): - iterable1 = iter([1, 2, 3]) - iterable2 = process_iterable_helper(iterable1, processor=processor) - self.assertIsInstance(iterable2, collections.abc.Iterable) - list2 = list(iterable2) - self.assertEqual(list2, [2, 4, 6]) - - def test_normal_combined(self): - iterable1 = iter([1, 2, 10, 3, 6, 18, 5, 15]) - iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1, - out_predicate=predicate2, processor=processor) - self.assertIsInstance(iterable2, collections.abc.Iterable) - list2 = list(iterable2) - self.assertEqual(list2, [2, 6]) - - -class ProcessIterableHelperAsyncTest(unittest.TestCase): - - @deferred_f_from_coro_f - async def test_async_in_predicate(self): - iterable1 = as_async_generator([1, 2, 3]) - iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1) - self.assertIsInstance(iterable2, collections.abc.AsyncIterable) - list2 = await collect_asyncgen(iterable2) - self.assertEqual(list2, [1, 3]) - - @deferred_f_from_coro_f - async def test_async_out_predicate(self): - iterable1 = as_async_generator([1, 2, 10, 3, 15]) - iterable2 = process_iterable_helper(iterable1, out_predicate=predicate2) - self.assertIsInstance(iterable2, collections.abc.AsyncIterable) - list2 = await collect_asyncgen(iterable2) - self.assertEqual(list2, [1, 2, 3]) - - @deferred_f_from_coro_f - async def test_async_processor(self): - iterable1 = as_async_generator([1, 2, 3]) - iterable2 = process_iterable_helper(iterable1, processor=processor) - self.assertIsInstance(iterable2, collections.abc.AsyncIterable) - list2 = await collect_asyncgen(iterable2) - self.assertEqual(list2, [2, 4, 6]) - - @deferred_f_from_coro_f - async def test_async_combined(self): - iterable1 = as_async_generator([1, 2, 10, 3, 6, 18, 5, 15]) - iterable2 = process_iterable_helper(iterable1, in_predicate=predicate1, - out_predicate=predicate2, processor=processor) - self.assertIsInstance(iterable2, collections.abc.AsyncIterable) - list2 = await collect_asyncgen(iterable2) - self.assertEqual(list2, [2, 6])