From 207fb6f36c0927548bf020eb4003dff417d3b6dd Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Wed, 18 Dec 2019 19:34:50 +0500 Subject: [PATCH] Initial support for async generator start_requests. --- conftest.py | 2 + docs/topics/coroutines.rst | 42 +++++++++- docs/topics/spider-middleware.rst | 13 +++- pytest.ini | 2 + scrapy/core/engine.py | 18 ++++- scrapy/crawler.py | 16 +++- scrapy/utils/asyncgen.py | 20 +++++ scrapy/utils/py36.py | 10 --- scrapy/utils/spider.py | 2 +- tests/py36/_test_engine.py | 19 +++++ tests/py36/_test_spidermiddleware.py | 8 ++ tests/py36/_test_utils_asyncgen.py | 3 + tests/test_engine.py | 27 +++++++ tests/test_spidermiddleware.py | 112 +++++++++++++++++++++++++++ tests/test_utils_asyncgen.py | 38 +++++++++ 15 files changed, 311 insertions(+), 21 deletions(-) create mode 100644 scrapy/utils/asyncgen.py delete mode 100644 scrapy/utils/py36.py create mode 100644 tests/py36/_test_engine.py create mode 100644 tests/py36/_test_spidermiddleware.py create mode 100644 tests/py36/_test_utils_asyncgen.py create mode 100644 tests/test_utils_asyncgen.py diff --git a/conftest.py b/conftest.py index be5fbabf4..67e6fd3fe 100644 --- a/conftest.py +++ b/conftest.py @@ -8,6 +8,8 @@ def _py_files(folder): collect_ignore = [ + # no tests and not importable on 3.5 + "scrapy/utils/asyncgen.py", # not a test, but looks like a test "scrapy/utils/testsite.py", # contains scripts to be run by tests/test_crawler.py::CrawlerProcessSubprocess diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst index 487cf4c6c..c91a7b42a 100644 --- a/docs/topics/coroutines.rst +++ b/docs/topics/coroutines.rst @@ -46,10 +46,48 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``): methods of :ref:`downloader middlewares `. +- The + :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_start_requests` + method of :ref:`spider middlewares `. + - :ref:`Signal handlers that support deferreds `. +- The :meth:`start_requests` spider method. + .. _asynchronous generators were introduced in Python 3.6: https://www.python.org/dev/peps/pep-0525/ +.. _async-start_requests: + +Asynchronous start_requests and spider middlewares +================================================== + +.. versionadded:: 2.1 + +The :meth:`start_requests` spider method can be an asynchronous generator. In +this case all spider middlewares used with this spider that have the +:meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_start_requests` +method must support this: if they receive an asynchronous iterable, they must +return one as well. On the other hand, if they receive a normal iterable, they +shouldn't break. Such universal :meth:`process_start_requests` must be an +asynchronous generator itself, and so it will always convert a normal iterable +to an asynchronous one. As a result of a middleware method is passed to the +same method of the next middleware, it's only possible to mix middlewares with +synchronous and asynchronous :meth:`process_start_requests` if all synchronous +ones are called first. + +Here is an example of a universal middleware:: + + from scrapy.utils.asyncgen import as_async_generator + + class ProcessStartRequestsAsyncGenMiddleware: + async def process_start_requests(self, start_requests, spider): + async for r in as_async_generator(start_requests): + # ... do something with r + yield r + +If this method includes asynchronous code, that code will work even with +synchronous :meth:`start_requests`. + Usage ===== @@ -98,8 +136,8 @@ many useful Python libraries providing such code:: Common use cases for asynchronous code include: -* requesting data from websites, databases and other services (in callbacks, - pipelines and middlewares); +* requesting data from websites, databases and other services (in + :meth:`start_requests`, callbacks, pipelines and middlewares); * storing data in databases (in pipelines and middlewares); * delaying the spider initialization until some external event (in the :signal:`spider_opened` handler); diff --git a/docs/topics/spider-middleware.rst b/docs/topics/spider-middleware.rst index 0e8210130..2c055835f 100644 --- a/docs/topics/spider-middleware.rst +++ b/docs/topics/spider-middleware.rst @@ -148,6 +148,9 @@ object gives you access, for example, to the :ref:`settings `. .. method:: process_start_requests(start_requests, spider) .. versionadded:: 0.15 + .. versionchanged:: 2.1 + Since 2.1 this can take and return an :term:`python:asynchronous + iterable`. This method is called with the start requests of the spider, and works similarly to the :meth:`process_spider_output` method, except that it @@ -157,6 +160,15 @@ object gives you access, for example, to the :ref:`settings `. It receives an iterable (in the ``start_requests`` parameter) and must return another iterable of :class:`~scrapy.http.Request` objects. + When using Python 3.6+ `start_requests` may be an + :term:`python:asynchronous iterable`. To support such spiders, this + middleware method should be an :term:`python:asynchronous generator`, + which should support both synchronous and asynchronous + ``start_requests``. :func:`scrapy.utils.asyncgen.as_async_generator` + can be used to convert both kinds of ``start_requests`` to an + asynchronous iterable. See :ref:`more details here + `. + .. note:: When implementing this method in your spider middleware, you should always return an iterable (that follows the input one) and not consume all ``start_requests`` iterator because it can be very @@ -186,7 +198,6 @@ object gives you access, for example, to the :ref:`settings `. .. _Exception: https://docs.python.org/2/library/exceptions.html#exceptions.Exception - .. _topics-spider-middleware-ref: Built-in spider middleware reference diff --git a/pytest.ini b/pytest.ini index 141a13a4f..446408548 100644 --- a/pytest.ini +++ b/pytest.ini @@ -119,6 +119,7 @@ flake8-ignore = scrapy/spiders/sitemap.py E501 # scrapy/utils scrapy/utils/asyncio.py E501 + scrapy/utils/asyncgen.py E501 scrapy/utils/benchserver.py E501 scrapy/utils/conf.py E402 E501 scrapy/utils/datatypes.py E501 @@ -229,6 +230,7 @@ flake8-ignore = tests/test_spidermiddleware_referer.py E501 F841 E125 E201 E124 E501 E241 E121 tests/test_squeues.py E501 E741 tests/test_utils_asyncio.py E501 + tests/test_utils_asyncgen.py E501 tests/test_utils_conf.py E501 E128 tests/test_utils_curl.py E501 tests/test_utils_datatypes.py E402 E501 diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index 74f03344e..8326bf9a3 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -4,6 +4,7 @@ This is the Scrapy engine which controls the Scheduler, Downloader and Spiders. For more information see docs/topics/architecture.rst """ +import inspect import logging from time import time @@ -14,6 +15,7 @@ from scrapy import signals from scrapy.core.scraper import Scraper from scrapy.exceptions import DontCloseSpider from scrapy.http import Response, Request +from scrapy.utils.defer import deferred_from_coro from scrapy.utils.misc import load_object from scrapy.utils.reactor import CallLaterOnce from scrapy.utils.log import logformatter_adapter, failure_to_exc_info @@ -26,7 +28,10 @@ class Slot: def __init__(self, start_requests, close_if_idle, nextcall, scheduler): self.closing = False self.inprogress = set() # requests in progress - self.start_requests = iter(start_requests) + if hasattr(inspect, 'isasyncgen') and inspect.isasyncgen(start_requests): + self.start_requests = start_requests + else: + self.start_requests = iter(start_requests) self.close_if_idle = close_if_idle self.nextcall = nextcall self.scheduler = scheduler @@ -110,6 +115,7 @@ class ExecutionEngine: """Resume the execution engine""" self.paused = False + @defer.inlineCallbacks def _next_request(self, spider): slot = self.slot if not slot: @@ -124,8 +130,14 @@ class ExecutionEngine: if slot.start_requests and not self._needs_backout(spider): try: - request = next(slot.start_requests) - except StopIteration: + if hasattr(inspect, 'isasyncgen') and inspect.isasyncgen(slot.start_requests): + request = yield deferred_from_coro(slot.start_requests.__anext__()) + else: + request = next(slot.start_requests) + except (StopIteration, StopAsyncIteration): + if slot.start_requests is None: + # the crawl is already stopping + return slot.start_requests = None except Exception: slot.start_requests = None diff --git a/scrapy/crawler.py b/scrapy/crawler.py index c982611fb..dc92e8223 100644 --- a/scrapy/crawler.py +++ b/scrapy/crawler.py @@ -86,10 +86,7 @@ class Crawler: try: self.spider = self._create_spider(*args, **kwargs) self.engine = self._create_engine() - if inspect.iscoroutinefunction(self.spider.start_requests): - start_requests = yield deferred_from_coro(self.spider.start_requests()) - else: - start_requests = iter(self.spider.start_requests()) + start_requests = yield self.call_start_requests() yield self.engine.open_spider(self.spider, start_requests) yield defer.maybeDeferred(self.engine.start) except Exception: @@ -98,6 +95,17 @@ class Crawler: yield self.engine.close() raise + @defer.inlineCallbacks + def call_start_requests(self): + if hasattr(inspect, 'isasyncgenfunction') and inspect.isasyncgenfunction(self.spider.start_requests): + # requires Python 3.6+ + start_requests = self.spider.start_requests().__aiter__() + elif inspect.iscoroutinefunction(self.spider.start_requests): + start_requests = yield deferred_from_coro(self.spider.start_requests()) + else: + start_requests = iter(self.spider.start_requests()) + return start_requests + def _create_spider(self, *args, **kwargs): return self.spidercls.from_crawler(self, *args, **kwargs) diff --git a/scrapy/utils/asyncgen.py b/scrapy/utils/asyncgen.py new file mode 100644 index 000000000..6130e81be --- /dev/null +++ b/scrapy/utils/asyncgen.py @@ -0,0 +1,20 @@ +""" +Helpers using Python 3.6+ async generator syntax (ignore SyntaxError on import). +""" +import collections + + +async def collect_asyncgen(result): + results = [] + async for x in result: + results.append(x) + return results + + +async def as_async_generator(it): + if isinstance(it, collections.abc.AsyncIterator): + async for r in it: + yield r + else: + for r in it: + yield r diff --git a/scrapy/utils/py36.py b/scrapy/utils/py36.py deleted file mode 100644 index c8c24076e..000000000 --- a/scrapy/utils/py36.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Helpers using Python 3.6+ syntax (ignore SyntaxError on import). -""" - - -async def collect_asyncgen(result): - results = [] - async for x in result: - results.append(x) - return results diff --git a/scrapy/utils/spider.py b/scrapy/utils/spider.py index 1b8a82829..59038ef66 100644 --- a/scrapy/utils/spider.py +++ b/scrapy/utils/spider.py @@ -5,7 +5,7 @@ from scrapy.spiders import Spider from scrapy.utils.defer import deferred_from_coro from scrapy.utils.misc import arg_to_iter try: - from scrapy.utils.py36 import collect_asyncgen + from scrapy.utils.asyncgen import collect_asyncgen except SyntaxError: collect_asyncgen = None diff --git a/tests/py36/_test_engine.py b/tests/py36/_test_engine.py new file mode 100644 index 000000000..d149bf482 --- /dev/null +++ b/tests/py36/_test_engine.py @@ -0,0 +1,19 @@ +#coding: utf-8 +import asyncio + +from scrapy import Request + +from tests.test_engine import TestSpider + + +class StartRequestsAsyncGenSpider(TestSpider): + async def start_requests(self): + for url in self.start_urls: + yield Request(url, dont_filter=True) + + +class StartRequestsAsyncGenAsyncioSpider(TestSpider): + async def start_requests(self): + for url in self.start_urls: + yield Request(url, dont_filter=True) + await asyncio.sleep(0.1) diff --git a/tests/py36/_test_spidermiddleware.py b/tests/py36/_test_spidermiddleware.py new file mode 100644 index 000000000..be49aadfb --- /dev/null +++ b/tests/py36/_test_spidermiddleware.py @@ -0,0 +1,8 @@ +# coding: utf-8 +from scrapy.utils.asyncgen import as_async_generator + + +class ProcessStartRequestsAsyncGenMiddleware: + async def process_start_requests(self, start_requests, spider): + async for r in as_async_generator(start_requests): + yield r diff --git a/tests/py36/_test_utils_asyncgen.py b/tests/py36/_test_utils_asyncgen.py new file mode 100644 index 000000000..d66e28487 --- /dev/null +++ b/tests/py36/_test_utils_asyncgen.py @@ -0,0 +1,3 @@ +async def async_gen(): + for i in range(3): + yield i diff --git a/tests/test_engine.py b/tests/test_engine.py index 0936dd7e6..6047ee3d0 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -15,6 +15,7 @@ import re import sys from urllib.parse import urlparse +from pytest import mark from twisted.internet import reactor, defer from twisted.web import server, static, util from twisted.trial import unittest @@ -212,6 +213,32 @@ class EngineTest(unittest.TestCase): yield self.run.run() self._assert_visited_urls() + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @defer.inlineCallbacks + def test_crawler_startrequests_asyncgen(self): + from tests.py36._test_engine import StartRequestsAsyncGenSpider + self.run = CrawlerRun(StartRequestsAsyncGenSpider) + yield self.run.run() + self._assert_visited_urls() + + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @mark.only_asyncio() + @defer.inlineCallbacks + def test_crawler_startrequests_asyncgen_asyncio(self): + from tests.py36._test_engine import StartRequestsAsyncGenAsyncioSpider + self.run = CrawlerRun(StartRequestsAsyncGenAsyncioSpider) + yield self.run.run() + self._assert_visited_urls() + + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @mark.only_asyncio() + @defer.inlineCallbacks + def test_crawler_startrequests_asyncgen_asyncio_mw1(self): + from tests.py36._test_engine import StartRequestsAsyncGenAsyncioSpider + self.run = CrawlerRun(StartRequestsAsyncGenAsyncioSpider) + yield self.run.run() + self._assert_visited_urls() + def _assert_visited_urls(self): must_be_visited = ["/", "/redirect", "/redirected", "/item1.html", "/item2.html", "/item999.html"] diff --git a/tests/test_spidermiddleware.py b/tests/test_spidermiddleware.py index 78e926adc..ce84ae187 100644 --- a/tests/test_spidermiddleware.py +++ b/tests/test_spidermiddleware.py @@ -1,13 +1,19 @@ +import inspect +import sys from unittest import mock +from pytest import mark +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.defer import deferred_from_coro from scrapy.utils.test import get_crawler from scrapy.core.spidermw import SpiderMiddlewareManager +from tests.test_engine import StartRequestsAsyncDefSpider class SpiderMiddlewareTestCase(TestCase): @@ -101,3 +107,109 @@ class ProcessSpiderExceptionReRaise(SpiderMiddlewareTestCase): result = self._scrape_response() self.assertIsInstance(result, Failure) self.assertIsInstance(result.value, ZeroDivisionError) + + +class ProcessStartRequestsSimpleMiddleware: + def process_start_requests(self, start_requests, spider): + for r in start_requests: + yield r + + +class ProcessStartRequestsSimple(TestCase): + """ process_start_requests tests for simple start_requests""" + + spider_cls = Spider + + @defer.inlineCallbacks + def _get_processed_start_requests(self, *mw_classes): + crawler = get_crawler(self.spider_cls) + start_urls = ['https://example.com/%d' % i for i in range(3)] + crawler.spider = crawler._create_spider('foo', start_urls=start_urls) + mwman = SpiderMiddlewareManager.from_crawler(crawler) + for mw_cls in mw_classes: + mwman._add_middleware(mw_cls()) + start_requests = yield crawler.call_start_requests() + processed_start_requests = yield mwman.process_start_requests(start_requests, crawler.spider) + return processed_start_requests + + def assertAsyncGeneratorNotIterable(self, processed_start_requests): + with self.assertRaisesRegex(TypeError, "'async_generator' object is not iterable"): + list(processed_start_requests) + + @defer.inlineCallbacks + def test_simple(self): + """ Simple mw """ + processed_start_requests = yield self._get_processed_start_requests(ProcessStartRequestsSimpleMiddleware) + self.assertTrue(inspect.isgenerator(processed_start_requests)) + start_requests_list = list(processed_start_requests) + self.assertEqual(len(start_requests_list), 3) + self.assertIsInstance(start_requests_list[0], Request) + + @defer.inlineCallbacks + def _test_asyncgen_base(self, *mw_classes): + from scrapy.utils.asyncgen import collect_asyncgen + processed_start_requests = yield self._get_processed_start_requests(*mw_classes) + self.assertTrue(inspect.isasyncgen(processed_start_requests)) + start_requests_list = yield deferred_from_coro(collect_asyncgen(processed_start_requests)) + self.assertEqual(len(start_requests_list), 3) + self.assertIsInstance(start_requests_list[0], Request) + + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @defer.inlineCallbacks + def test_asyncgen(self): + """ Asyncgen mw """ + from tests.py36._test_spidermiddleware import ProcessStartRequestsAsyncGenMiddleware + yield self._test_asyncgen_base(ProcessStartRequestsAsyncGenMiddleware) + + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @defer.inlineCallbacks + def test_asyncgen2(self): + """ Simple mw -> asyncgen mw """ + from tests.py36._test_spidermiddleware import ProcessStartRequestsAsyncGenMiddleware + yield self._test_asyncgen_base(ProcessStartRequestsAsyncGenMiddleware, + ProcessStartRequestsSimpleMiddleware) + + @mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") + @defer.inlineCallbacks + def test_asyncgen3(self): + """ Asyncgen mw -> simple mw; cannot work """ + from tests.py36._test_spidermiddleware import ProcessStartRequestsAsyncGenMiddleware + processed_start_requests = yield self._get_processed_start_requests( + ProcessStartRequestsSimpleMiddleware, + ProcessStartRequestsAsyncGenMiddleware) + self.assertTrue(inspect.isgenerator(processed_start_requests)) + self.assertAsyncGeneratorNotIterable(processed_start_requests) + + +class ProcessStartRequestsAsyncDef(ProcessStartRequestsSimple): + """ process_start_requests tests for async def start_requests """ + + spider_cls = StartRequestsAsyncDefSpider + + +@mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") +class ProcessStartRequestsAsyncGen(ProcessStartRequestsSimple): + """ process_start_requests tests for async generator start_requests """ + + def __init__(self, methodName='runTest'): + super().__init__(methodName) + from tests.py36._test_engine import StartRequestsAsyncGenSpider + self.spider_cls = StartRequestsAsyncGenSpider + + @defer.inlineCallbacks + def test_simple(self): + """ Simple mw; cannot work """ + processed_start_requests = yield self._get_processed_start_requests( + ProcessStartRequestsSimpleMiddleware) + self.assertTrue(inspect.isgenerator(processed_start_requests)) + self.assertAsyncGeneratorNotIterable(processed_start_requests) + + @defer.inlineCallbacks + def test_asyncgen2(self): + """ Simple mw -> asyncgen mw; cannot work """ + from tests.py36._test_spidermiddleware import ProcessStartRequestsAsyncGenMiddleware + processed_start_requests = yield self._get_processed_start_requests( + ProcessStartRequestsAsyncGenMiddleware, + ProcessStartRequestsSimpleMiddleware) + self.assertTrue(inspect.isasyncgen(processed_start_requests)) + self.assertAsyncGeneratorNotIterable(processed_start_requests) diff --git a/tests/test_utils_asyncgen.py b/tests/test_utils_asyncgen.py new file mode 100644 index 000000000..7f9c0371b --- /dev/null +++ b/tests/test_utils_asyncgen.py @@ -0,0 +1,38 @@ +import sys +from unittest import TestCase + +from pytest import mark + + +@mark.skipif(sys.version_info < (3, 6), reason="Async generators require Python 3.6 or higher") +class AsyncGeneratorTest(TestCase): + + async def test_as_async_generator_simple(self): + from scrapy.utils.asyncgen import as_async_generator + gen = (i for i in range(3)) + results = [] + async for i in as_async_generator(gen): + results.append(i) + self.assertEqual(results, [1, 2, 3]) + + async def test_as_async_generator_list(self): + from scrapy.utils.asyncgen import as_async_generator + L = [i for i in range(3)] + results = [] + async for i in as_async_generator(L): + results.append(i) + self.assertEqual(results, [1, 2, 3]) + + async def test_as_async_generator_async(self): + from scrapy.utils.asyncgen import as_async_generator + from tests.py36._test_utils_asyncgen import async_gen + results = [] + async for i in as_async_generator(async_gen()): + results.append(i) + self.assertEqual(results, [1, 2, 3]) + + async def test_collect_asyncgen(self): + from scrapy.utils.asyncgen import collect_asyncgen + from tests.py36._test_utils_asyncgen import async_gen + results = collect_asyncgen(async_gen()) + self.assertEqual(results, [1, 2, 3])