RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Drop process_iterable_helper, add _process_iterable_universal.

This commit is contained in:
Andrey Rakhmatullin 2021-02-26 19:32:45 +05:00
parent f9a5385146
commit c51ec1ae1c
9 changed files with 120 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
scrapy/core/spidermw.py
View File

@ -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

18
scrapy/spidermiddlewares/depth.py
View File

@ -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)

32
scrapy/spidermiddlewares/offsite.py
View File

@ -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

9
scrapy/spidermiddlewares/referer.py
View File

@ -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

9
scrapy/spidermiddlewares/urllength.py
View File

@ -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)

45
scrapy/utils/asyncgen.py
View File

@ -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

35
scrapy/utils/middlewares.py
View File

@ -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)

22
tests/test_utils_asyncgen.py
View File

@ -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])

87
tests/test_utils_middlewares.py
View File

@ -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])