diff --git a/docs/intro/tutorial.rst b/docs/intro/tutorial.rst index 71b420be6..c4e04364b 100644 --- a/docs/intro/tutorial.rst +++ b/docs/intro/tutorial.rst @@ -172,7 +172,7 @@ request (in this case, the ``parse`` method) with a A shortcut to the ``start`` method ----------------------------------------- +---------------------------------- Instead of implementing a :meth:`~scrapy.Spider.start` method that yields :class:`~scrapy.Request` objects from URLs, you can define a diff --git a/docs/news.rst b/docs/news.rst index 27a06358f..0add34a81 100644 --- a/docs/news.rst +++ b/docs/news.rst @@ -22,14 +22,19 @@ Backward-incompatible changes As a result, the order in which start requests are sent may change. See :ref:`start-requests` for details and information on how to force start - request order or pause start request iteration while there are scheduled - requests. + request order or :ref:`pause start request iteration while there are + scheduled requests `. + +- An unhandled exception from the + :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.open_spider` method of a + :ref:`spider middleware ` no longer stops the + crawl. - In ``scrapy.core.engine.ExecutionEngine``: - - The second parameter of ``open_spider()``, ``start_requests()``, has - been removed. The starting requests are determined by the ``spider`` - parameter instead (see :meth:`~scrapy.Spider.start`). + - The second parameter of ``open_spider()``, ``start_requests``, has been + removed. The start requests are determined by the ``spider`` parameter + instead (see :meth:`~scrapy.Spider.start`). - The ``slot`` attribute has been renamed to ``_slot`` and should not be used. @@ -68,7 +73,8 @@ New features - You can now yield the start requests and items of a spider from the :meth:`~scrapy.Spider.start` spider method and from the :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_start` spider - middleware method, both asynchronous generators. + middleware method, both :term:`asynchronous generators `. This makes it possible to use asynchronous code to generate those start requests and items, e.g. reading them from a queue service or database @@ -84,6 +90,15 @@ New features (:issue:`456`, :issue:`3477`, :issue:`4467`, :issue:`5627`, :issue:`6729`) +- :class:`Crawler.signals ` has a new + :meth:`~scrapy.signalmanager.SignalManager.wait_for` method. + +- Added a new :signal:`scheduler_empty` signal. + +- Exposed a new method of :class:`Crawler.engine + `: + :meth:`~scrapy.core.engine.ExecutionEngine.needs_backout`. + Bug fixes ~~~~~~~~~ @@ -398,9 +413,9 @@ New features - ``scrapy.Spider.start_requests()`` can now yield items. (:issue:`5289`, :issue:`6417`) - .. note:: Some third-party spider middlewares may need to be updated for - Scrapy 2.12 support before you can use them in combination with the - ability to yield items from ``start_requests()``. + .. note:: Some spider middlewares may need to be updated for Scrapy 2.12 + support before you can use them in combination with the ability to + yield items from ``start_requests()``. - Added a new :class:`~scrapy.http.Response` subclass, :class:`~scrapy.http.JsonResponse`, for responses with a `JSON MIME type diff --git a/docs/topics/api.rst b/docs/topics/api.rst index 5a00fd570..8e8f3a0c9 100644 --- a/docs/topics/api.rst +++ b/docs/topics/api.rst @@ -280,3 +280,9 @@ class (which they all inherit from). Close the given spider. After this is called, no more specific stats can be accessed or collected. + +Engine API +========== + +.. autoclass:: scrapy.core.engine.ExecutionEngine() + :members: needs_backout diff --git a/docs/topics/coroutines.rst b/docs/topics/coroutines.rst index aae2950b6..18aa8c68a 100644 --- a/docs/topics/coroutines.rst +++ b/docs/topics/coroutines.rst @@ -18,7 +18,8 @@ Supported callables The following callables may be defined as coroutines using ``async def``, and hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``): -- The :meth:`~scrapy.spiders.Spider.start` spider method. +- The :meth:`~scrapy.spiders.Spider.start` spider method, which *must* be + defined as an :term:`asynchronous generator`. .. versionadded: VERSION @@ -46,8 +47,8 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``): :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output` method of :ref:`spider middlewares `. - It must be defined as an :term:`asynchronous generator`. The input - ``result`` parameter is an :term:`asynchronous iterable`. + If defined as a coroutine, it must be 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`. @@ -55,7 +56,8 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``): .. versionadded:: 2.7 - The :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_start` method - of :ref:`spider middlewares `. + of :ref:`spider middlewares `, which *must* be + defined as an :term:`asynchronous generator`. .. versionadded:: VERSION diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst index b13e2950f..66cb87fc5 100644 --- a/docs/topics/signals.rst +++ b/docs/topics/signals.rst @@ -142,6 +142,8 @@ scheduler_empty :meth:`~scrapy.core.scheduler.BaseScheduler.next_request` method) and the scheduler returns none. + See :ref:`start-requests-lazy` for an example. + Item signals ------------ diff --git a/docs/topics/spider-middleware.rst b/docs/topics/spider-middleware.rst index 0baf094bc..5b523893a 100644 --- a/docs/topics/spider-middleware.rst +++ b/docs/topics/spider-middleware.rst @@ -70,7 +70,7 @@ one or more of these methods: .. class:: SpiderMiddleware - .. method:: process_start(start: AsyncIterable[Any], /) -> AsyncIterable[Any] + .. method:: process_start(start: AsyncIterator[Any], /) -> AsyncIterator[Any] :async: Iterate over the output of :meth:`~scrapy.Spider.start` or that diff --git a/docs/topics/spiders.rst b/docs/topics/spiders.rst index c640609cc..ae80f347d 100644 --- a/docs/topics/spiders.rst +++ b/docs/topics/spiders.rst @@ -373,6 +373,35 @@ Scrapy does not try to send :meth:`~scrapy.Spider.start` requests in order. Instead, it prioritizes reaching :setting:`CONCURRENT_REQUESTS` and :ref:`scheduling ` start requests. +.. + The request send order when all start requests and callback requests have + the same priority is rather unintuitive: + + 1. First, the first CONCURRENT_REQUESTS start requests are sent in order. + + Awaiting slow operations in Spider.start() can lower that. + + 2. Then, assuming an even domain distribution in start requests (i.e. + ABCABC, not AABBCC), the last N start requests are sent in reverse + order, where N is: + + min(CONCURRENT_REQUESTS, CONCURRENT_REQUESTS_PER_DOMAIN * domain_count) + + 3. Finally, the remaining start requests are also sent in reverse order, + but only when there are not enough pending requests yielded from + callbacks to reach the configured concurrency. + + The reverse order is because the scheduler uses a LIFO queue by default + (SCHEDULER_MEMORY_QUEUE, SCHEDULER_DISK_QUEUE). The order of the first few + requests is unnaffected because they are sent as soon as they are + scheduled. The last start requests sent before callback requests are those + that can be sent before the first callback requests are scheduled. + + We do not document this behavior, so that we may change it in the future + without breaking the contract. + +.. _start-requests-order: + Forcing a start request order ----------------------------- @@ -405,6 +434,8 @@ To force a specific **request order**, override the You can also :ref:`customize the scheduler ` if you need more control over request prioritization. +.. _start-requests-lazy: + Delaying start request iteration -------------------------------- diff --git a/scrapy/commands/parse.py b/scrapy/commands/parse.py index a43f55137..0dd9954cb 100644 --- a/scrapy/commands/parse.py +++ b/scrapy/commands/parse.py @@ -22,7 +22,7 @@ from scrapy.utils.spider import spidercls_for_request if TYPE_CHECKING: import argparse - from collections.abc import AsyncGenerator, AsyncIterable, Coroutine, Iterable + from collections.abc import AsyncGenerator, AsyncIterator, Coroutine, Iterable from twisted.python.failure import Failure @@ -258,7 +258,7 @@ class Command(BaseRunSpiderCommand): if not self.spidercls: logger.error("Unable to find spider for: %(url)s", {"url": url}) - async def start(spider: Spider) -> AsyncIterable[Any]: + async def start(spider: Spider) -> AsyncIterator[Any]: yield self.prepare_request(spider, Request(url), opts) if self.spidercls: diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index 4ef996464..1bbf8921f 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -241,6 +241,11 @@ class ExecutionEngine: self._spider_idle() def needs_backout(self) -> bool: + """Returns ``True`` if no more requests can be sent at the moment, or + ``False`` otherwise. + + See :ref:`start-requests-lazy` for an example. + """ assert self._slot is not None # typing assert self.scraper.slot is not None # typing return ( diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index 496adb500..85dd9b559 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -5,7 +5,7 @@ from __future__ import annotations import logging from collections import deque -from collections.abc import AsyncIterable, Iterator +from collections.abc import AsyncIterator, Iterator from typing import TYPE_CHECKING, Any, TypeVar, Union, cast from twisted.internet.defer import Deferred, inlineCallbacks @@ -170,7 +170,7 @@ class Scraper: raise TypeError( f"Incorrect type: expected Response or Failure, got {type(result)}: {result!r}" ) - dfd: Deferred[Iterable[Any] | AsyncIterable[Any]] = self._scrape2( + dfd: Deferred[Iterable[Any] | AsyncIterator[Any]] = self._scrape2( result, request, spider ) # returns spider's processed output dfd.addErrback(self.handle_spider_error, request, result, spider) @@ -181,7 +181,7 @@ class Scraper: def _scrape2( self, result: Response | Failure, request: Request, spider: Spider - ) -> Deferred[Iterable[Any] | AsyncIterable[Any]]: + ) -> Deferred[Iterable[Any] | AsyncIterator[Any]]: """ Handle the different cases of request's result been a Response or a Failure """ @@ -197,7 +197,7 @@ class Scraper: def call_spider( self, result: Response | Failure, request: Request, spider: Spider - ) -> Deferred[Iterable[Any] | AsyncIterable[Any]]: + ) -> Deferred[Iterable[Any] | AsyncIterator[Any]]: dfd: Deferred[Any] if isinstance(result, Response): if getattr(result, "request", None) is None: @@ -216,7 +216,7 @@ class Scraper: if request.errback: warn_on_generator_with_return_value(spider, request.errback) dfd.addErrback(request.errback) - dfd2: Deferred[Iterable[Any] | AsyncIterable[Any]] = dfd.addCallback( + dfd2: Deferred[Iterable[Any] | AsyncIterator[Any]] = dfd.addCallback( iterate_spider_output ) return dfd2 @@ -253,16 +253,16 @@ class Scraper: def handle_spider_output( self, - result: Iterable[_T] | AsyncIterable[_T], + result: Iterable[_T] | AsyncIterator[_T], request: Request, response: Response, spider: Spider, ) -> _HandleOutputDeferred: if not result: return defer_succeed(None) - it: Iterable[_T] | AsyncIterable[_T] + it: Iterable[_T] | AsyncIterator[_T] dfd: Deferred[_ParallelResult] - if isinstance(result, AsyncIterable): + if isinstance(result, AsyncIterator): it = aiter_errback( result, self.handle_spider_error, request, response, spider ) diff --git a/scrapy/core/spidermw.py b/scrapy/core/spidermw.py index f8946b221..ec125bb3e 100644 --- a/scrapy/core/spidermw.py +++ b/scrapy/core/spidermw.py @@ -7,7 +7,7 @@ See documentation in docs/topics/spider-middleware.rst from __future__ import annotations import logging -from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable +from collections.abc import AsyncIterator, Callable, Iterable from inspect import isasyncgenfunction, iscoroutine from itertools import islice from typing import TYPE_CHECKING, Any, TypeVar, Union, cast @@ -41,12 +41,12 @@ logger = logging.getLogger(__name__) _T = TypeVar("_T") ScrapeFunc = Callable[ - [Union[Response, Failure], Request, Spider], Union[Iterable[_T], AsyncIterable[_T]] + [Union[Response, Failure], Request, Spider], Union[Iterable[_T], AsyncIterator[_T]] ] def _isiterable(o: Any) -> bool: - return isinstance(o, (Iterable, AsyncIterable)) + return isinstance(o, (Iterable, AsyncIterator)) class SpiderMiddlewareManager(MiddlewareManager): @@ -136,7 +136,7 @@ class SpiderMiddlewareManager(MiddlewareManager): response: Response, request: Request, spider: Spider, - ) -> Iterable[_T] | AsyncIterable[_T]: + ) -> Iterable[_T] | AsyncIterator[_T]: for method in self.methods["process_spider_input"]: method = cast(Callable, method) try: @@ -157,10 +157,10 @@ class SpiderMiddlewareManager(MiddlewareManager): self, response: Response, spider: Spider, - iterable: Iterable[_T] | AsyncIterable[_T], + iterable: Iterable[_T] | AsyncIterator[_T], exception_processor_index: int, recover_to: MutableChain[_T] | MutableAsyncChain[_T], - ) -> Iterable[_T] | AsyncIterable[_T]: + ) -> Iterable[_T] | AsyncIterator[_T]: def process_sync(iterable: Iterable[_T]) -> Iterable[_T]: try: yield from iterable @@ -176,7 +176,7 @@ class SpiderMiddlewareManager(MiddlewareManager): assert isinstance(recover_to, MutableChain) recover_to.extend(exception_result) - async def process_async(iterable: AsyncIterable[_T]) -> AsyncIterable[_T]: + async def process_async(iterable: AsyncIterator[_T]) -> AsyncIterator[_T]: try: async for r in iterable: yield r @@ -192,7 +192,7 @@ class SpiderMiddlewareManager(MiddlewareManager): assert isinstance(recover_to, MutableAsyncChain) recover_to.extend(exception_result) - if isinstance(iterable, AsyncIterable): + if isinstance(iterable, AsyncIterator): return process_async(iterable) return process_sync(iterable) @@ -251,13 +251,13 @@ class SpiderMiddlewareManager(MiddlewareManager): self, response: Response, spider: Spider, - result: Iterable[_T] | AsyncIterable[_T], + result: Iterable[_T] | AsyncIterator[_T], start_index: int = 0, ) -> Generator[Deferred[Any], Any, MutableChain[_T] | MutableAsyncChain[_T]]: # 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[_T] | MutableAsyncChain[_T] - last_result_is_async = isinstance(result, AsyncIterable) + last_result_is_async = isinstance(result, AsyncIterator) recovered = MutableAsyncChain() if last_result_is_async else MutableChain() # There are three cases for the middleware: def foo, async def foo, def foo + async def foo_async. @@ -284,7 +284,7 @@ class SpiderMiddlewareManager(MiddlewareManager): need_downgrade = True try: if need_upgrade: - # Iterable -> AsyncIterable + # Iterable -> AsyncIterator result = as_async_generator(result) elif need_downgrade: logger.warning( @@ -294,10 +294,10 @@ class SpiderMiddlewareManager(MiddlewareManager): f" https://docs.scrapy.org/en/latest/topics/coroutines.html#for-middleware-users" f" for more information." ) - assert isinstance(result, AsyncIterable) - # AsyncIterable -> Iterable + assert isinstance(result, AsyncIterator) + # AsyncIterator -> Iterable result = yield deferred_from_coro(collect_asyncgen(result)) - if isinstance(recovered, AsyncIterable): + if isinstance(recovered, AsyncIterator): recovered_collected = yield deferred_from_coro( collect_asyncgen(recovered) ) @@ -330,7 +330,7 @@ class SpiderMiddlewareManager(MiddlewareManager): f"{type(result)}" ) raise _InvalidOutput(msg) - last_result_is_async = isinstance(result, AsyncIterable) + last_result_is_async = isinstance(result, AsyncIterator) if last_result_is_async: return MutableAsyncChain(result, recovered) @@ -340,23 +340,23 @@ class SpiderMiddlewareManager(MiddlewareManager): self, response: Response, spider: Spider, - result: Iterable[_T] | AsyncIterable[_T], + result: Iterable[_T] | AsyncIterator[_T], ) -> MutableChain[_T] | MutableAsyncChain[_T]: recovered: MutableChain[_T] | MutableAsyncChain[_T] - if isinstance(result, AsyncIterable): + if isinstance(result, AsyncIterator): recovered = MutableAsyncChain() else: recovered = MutableChain() result = self._evaluate_iterable(response, spider, result, 0, recovered) result = await maybe_deferred_to_future( cast( - "Deferred[Iterable[_T] | AsyncIterable[_T]]", + "Deferred[Iterable[_T] | AsyncIterator[_T]]", self._process_spider_output(response, spider, result), ) ) - if isinstance(result, AsyncIterable): + if isinstance(result, AsyncIterator): return MutableAsyncChain(result, recovered) - if isinstance(recovered, AsyncIterable): + if isinstance(recovered, AsyncIterator): recovered_collected = await collect_asyncgen(recovered) recovered = MutableChain(recovered_collected) return MutableChain(result, recovered) @@ -369,7 +369,7 @@ class SpiderMiddlewareManager(MiddlewareManager): spider: Spider, ) -> Deferred[MutableChain[_T] | MutableAsyncChain[_T]]: async def process_callback_output( - result: Iterable[_T] | AsyncIterable[_T], + result: Iterable[_T] | AsyncIterator[_T], ) -> MutableChain[_T] | MutableAsyncChain[_T]: return await self._process_callback_output(response, spider, result) @@ -378,7 +378,7 @@ class SpiderMiddlewareManager(MiddlewareManager): ) -> Failure | MutableChain[_T] | MutableAsyncChain[_T]: return self._process_spider_exception(response, spider, _failure) - dfd: Deferred[Iterable[_T] | AsyncIterable[_T]] = mustbe_deferred( + dfd: Deferred[Iterable[_T] | AsyncIterator[_T]] = mustbe_deferred( self._process_spider_input, scrape_func, response, request, spider ) dfd2: Deferred[MutableChain[_T] | MutableAsyncChain[_T]] = dfd.addCallback( diff --git a/scrapy/signalmanager.py b/scrapy/signalmanager.py index 5bdc1b690..f8c50b5e3 100644 --- a/scrapy/signalmanager.py +++ b/scrapy/signalmanager.py @@ -76,7 +76,10 @@ class SignalManager: _signal.disconnect_all(signal, **kwargs) async def wait_for(self, signal): - """Await the next *signal*.""" + """Await the next *signal*. + + See :ref:`start-requests-lazy` for an example. + """ d = Deferred() def handle(): diff --git a/scrapy/spidermiddlewares/depth.py b/scrapy/spidermiddlewares/depth.py index 3164c1c03..5760411c0 100644 --- a/scrapy/spidermiddlewares/depth.py +++ b/scrapy/spidermiddlewares/depth.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Any from scrapy.http import Request, Response if TYPE_CHECKING: - from collections.abc import AsyncIterable, Iterable + from collections.abc import AsyncIterator, Iterable # typing.Self requires Python 3.11 from typing_extensions import Self @@ -54,8 +54,8 @@ class DepthMiddleware: return (r for r in result if self._filter(r, response, spider)) async def process_spider_output_async( - self, response: Response, result: AsyncIterable[Any], spider: Spider - ) -> AsyncIterable[Any]: + self, response: Response, result: AsyncIterator[Any], spider: Spider + ) -> AsyncIterator[Any]: self._init_depth(response, spider) async for r in result: if self._filter(r, response, spider): diff --git a/scrapy/spidermiddlewares/offsite.py b/scrapy/spidermiddlewares/offsite.py index 646beb911..060be1975 100644 --- a/scrapy/spidermiddlewares/offsite.py +++ b/scrapy/spidermiddlewares/offsite.py @@ -23,7 +23,7 @@ warnings.warn( ) if TYPE_CHECKING: - from collections.abc import AsyncIterable, Iterable + from collections.abc import AsyncIterator, Iterable # typing.Self requires Python 3.11 from typing_extensions import Self @@ -52,8 +52,8 @@ class OffsiteMiddleware: return (r for r in result if self._filter(r, spider)) async def process_spider_output_async( - self, response: Response, result: AsyncIterable[Any], spider: Spider - ) -> AsyncIterable[Any]: + self, response: Response, result: AsyncIterator[Any], spider: Spider + ) -> AsyncIterator[Any]: async for r in result: if self._filter(r, spider): yield r diff --git a/scrapy/spidermiddlewares/referer.py b/scrapy/spidermiddlewares/referer.py index a3a1e5b92..c824579b0 100644 --- a/scrapy/spidermiddlewares/referer.py +++ b/scrapy/spidermiddlewares/referer.py @@ -19,7 +19,7 @@ from scrapy.utils.python import to_unicode from scrapy.utils.url import strip_url if TYPE_CHECKING: - from collections.abc import AsyncIterable, Iterable + from collections.abc import AsyncIterator, Iterable # typing.Self requires Python 3.11 from typing_extensions import Self @@ -376,8 +376,8 @@ class RefererMiddleware: return (self._set_referer(r, response) for r in result) async def process_spider_output_async( - self, response: Response, result: AsyncIterable[Any], spider: Spider - ) -> AsyncIterable[Any]: + self, response: Response, result: AsyncIterator[Any], spider: Spider + ) -> AsyncIterator[Any]: async for r in result: yield self._set_referer(r, response) diff --git a/scrapy/spidermiddlewares/urllength.py b/scrapy/spidermiddlewares/urllength.py index a1cd1bb7c..9259fea3c 100644 --- a/scrapy/spidermiddlewares/urllength.py +++ b/scrapy/spidermiddlewares/urllength.py @@ -14,7 +14,7 @@ from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning from scrapy.http import Request, Response if TYPE_CHECKING: - from collections.abc import AsyncIterable, Iterable + from collections.abc import AsyncIterator, Iterable # typing.Self requires Python 3.11 from typing_extensions import Self @@ -57,8 +57,8 @@ class UrlLengthMiddleware: return (r for r in result if self._filter(r, spider)) async def process_spider_output_async( - self, response: Response, result: AsyncIterable[Any], spider: Spider - ) -> AsyncIterable[Any]: + self, response: Response, result: AsyncIterator[Any], spider: Spider + ) -> AsyncIterator[Any]: async for r in result: if self._filter(r, spider): yield r diff --git a/scrapy/spiders/crawl.py b/scrapy/spiders/crawl.py index 087049425..171d8479c 100644 --- a/scrapy/spiders/crawl.py +++ b/scrapy/spiders/crawl.py @@ -8,7 +8,7 @@ See documentation in docs/topics/spiders.rst from __future__ import annotations import copy -from collections.abc import AsyncIterable, Awaitable, Callable +from collections.abc import AsyncIterator, Awaitable, Callable from typing import TYPE_CHECKING, Any, Optional, TypeVar, cast from twisted.python.failure import Failure @@ -156,10 +156,10 @@ class CrawlSpider(Spider): callback: CallbackT | None, cb_kwargs: dict[str, Any], follow: bool = True, - ) -> AsyncIterable[Any]: + ) -> AsyncIterator[Any]: if callback: cb_res = callback(response, **cb_kwargs) or () - if isinstance(cb_res, AsyncIterable): + if isinstance(cb_res, AsyncIterator): cb_res = await collect_asyncgen(cb_res) elif isinstance(cb_res, Awaitable): cb_res = await cb_res diff --git a/scrapy/utils/asyncgen.py b/scrapy/utils/asyncgen.py index 237bd8331..6d96a41f5 100644 --- a/scrapy/utils/asyncgen.py +++ b/scrapy/utils/asyncgen.py @@ -1,20 +1,20 @@ from __future__ import annotations -from collections.abc import AsyncGenerator, AsyncIterable, Iterable +from collections.abc import AsyncGenerator, AsyncIterator, Iterable from typing import TypeVar _T = TypeVar("_T") -async def collect_asyncgen(result: AsyncIterable[_T]) -> list[_T]: +async def collect_asyncgen(result: AsyncIterator[_T]) -> list[_T]: return [x async for x in result] async def as_async_generator( - it: Iterable[_T] | AsyncIterable[_T], + it: Iterable[_T] | AsyncIterator[_T], ) -> AsyncGenerator[_T]: """Wraps an iterable (sync or async) into an async generator.""" - if isinstance(it, AsyncIterable): + if isinstance(it, AsyncIterator): async for r in it: yield r else: diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index 42ad28d8d..c39837716 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -22,7 +22,7 @@ from scrapy.exceptions import IgnoreRequest, ScrapyDeprecationWarning from scrapy.utils.reactor import _get_asyncio_event_loop, is_asyncio_reactor_installed if TYPE_CHECKING: - from collections.abc import AsyncIterable, AsyncIterator, Callable + from collections.abc import AsyncIterator, Callable from twisted.python.failure import Failure @@ -177,7 +177,7 @@ class _AsyncCooperatorAdapter(Iterator, Generic[_T]): def __init__( self, - aiterable: AsyncIterable[_T], + aiterable: AsyncIterator[_T], callable: Callable[Concatenate[_T, _P], Deferred[Any] | None], *callable_args: _P.args, **callable_kwargs: _P.kwargs, @@ -234,7 +234,7 @@ class _AsyncCooperatorAdapter(Iterator, Generic[_T]): def parallel_async( - async_iterable: AsyncIterable[_T], + async_iterable: AsyncIterator[_T], count: int, callable: Callable[Concatenate[_T, _P], Deferred[Any] | None], *args: _P.args, @@ -332,11 +332,11 @@ def iter_errback( async def aiter_errback( - aiterable: AsyncIterable[_T], + aiterable: AsyncIterator[_T], errback: Callable[Concatenate[Failure, _P], Any], *a: _P.args, **kw: _P.kwargs, -) -> AsyncIterable[_T]: +) -> AsyncIterator[_T]: """Wraps an async iterable calling an errback if an error is caught while iterating it. Similar to scrapy.utils.defer.iter_errback() """ diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py index 2e6869779..7e7b83c59 100644 --- a/scrapy/utils/python.py +++ b/scrapy/utils/python.py @@ -10,7 +10,7 @@ import re import sys import warnings import weakref -from collections.abc import AsyncIterable, Iterable, Mapping +from collections.abc import AsyncIterator, Iterable, Mapping from functools import partial, wraps from itertools import chain from typing import TYPE_CHECKING, Any, TypeVar, overload @@ -19,8 +19,9 @@ from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.utils.asyncgen import as_async_generator if TYPE_CHECKING: - from collections.abc import AsyncIterator, Callable, Iterator + from collections.abc import Callable, Iterator from re import Pattern + from typing import Self # typing.Concatenate and typing.ParamSpec require Python 3.10 from typing_extensions import Concatenate, ParamSpec @@ -369,25 +370,25 @@ class MutableChain(Iterable[_T]): async def _async_chain( - *iterables: Iterable[_T] | AsyncIterable[_T], + *iterables: Iterable[_T] | AsyncIterator[_T], ) -> AsyncIterator[_T]: for it in iterables: async for o in as_async_generator(it): yield o -class MutableAsyncChain(AsyncIterable[_T]): +class MutableAsyncChain(AsyncIterator[_T]): """ Similar to MutableChain but for async iterables """ - def __init__(self, *args: Iterable[_T] | AsyncIterable[_T]): + def __init__(self, *args: Iterable[_T] | AsyncIterator[_T]): self.data: AsyncIterator[_T] = _async_chain(*args) - def extend(self, *iterables: Iterable[_T] | AsyncIterable[_T]) -> None: + def extend(self, *iterables: Iterable[_T] | AsyncIterator[_T]) -> None: self.data = _async_chain(self.data, _async_chain(*iterables)) - def __aiter__(self) -> AsyncIterator[_T]: + def __aiter__(self) -> Self: return self async def __anext__(self) -> _T: diff --git a/tests/test_engine_loop.py b/tests/test_engine_loop.py index 8ef62d018..c1bca8f9a 100644 --- a/tests/test_engine_loop.py +++ b/tests/test_engine_loop.py @@ -41,7 +41,7 @@ class MainTestCase(TestCase): self.crawler.engine._slot.scheduler.pause() self.crawler.engine._slot.scheduler.enqueue_request(Request("data:,b")) - # During this time, the reactor reports having requests but + # During this time, the scheduler reports having requests but # returns None. await sleep(seconds) @@ -81,33 +81,6 @@ class MainTestCase(TestCase): class RequestSendOrderTestCase(TestCase): - """Test the intrincacies of request send order when all start requests and - callback requests have the same priority. - - It is a very unintuitive behavior, documented as “undefined” so that we may - change it in the future without breaking the contract: - - 1. First, the first CONCURRENT_REQUESTS start requests are sent in order. - - Awaiting slow operations in Spider.start() can lower that. - - 2. Then, assuming an even domain distribution in start requests (i.e. - ABCABC, not AABBCC), the last N start requests are sent in reverse - order, where N is: - - min(CONCURRENT_REQUESTS, CONCURRENT_REQUESTS_PER_DOMAIN * domain_count) - - 3. Finally, the remaining start requests are also sent in reverse order, - but only when there are not enough pending requests yielded from - callbacks to reach the configured concurrency. - - The reverse order is because the scheduler uses a LIFO queue by default - (SCHEDULER_MEMORY_QUEUE, SCHEDULER_DISK_QUEUE). The order of the first few - requests is unnaffected because they are sent as soon as they are - scheduled. The last start requests sent before callback requests are those - that can be sent before the first callback requests are scheduled. - """ - seconds = 0.1 # increase if flaky @classmethod @@ -238,9 +211,8 @@ class RequestSendOrderTestCase(TestCase): cb_nums=cb_nums, settings={ "CONCURRENT_REQUESTS": 1, - # Without the lazy approach, using the FIFO queue would - # yield a different result, with start requests not being - # sorted. + # Without the lazy approach, a FIFO queue would yield the + # start requests in a different order. "SCHEDULER_MEMORY_QUEUE": "scrapy.squeues.FifoMemoryQueue", }, response_seconds=response_seconds, diff --git a/tests/test_spider_start.py b/tests/test_spider_start.py index 006701220..b5eaa93f5 100644 --- a/tests/test_spider_start.py +++ b/tests/test_spider_start.py @@ -6,17 +6,11 @@ from twisted.internet.defer import Deferred from twisted.trial.unittest import TestCase from scrapy import Spider, signals -from scrapy.core.engine import ExecutionEngine from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future from scrapy.utils.test import get_crawler -# These are the minimum seconds necessary to wait to reproduce the issue that -# has been solved by catching the RuntimeError exception in the -# ExecutionEngine._next_request() method. A lower value makes these tests pass -# even if we remove that exception handling, but they start failing with this -# much delay. -ASYNC_GEN_ERROR_MINIMUM_SECONDS = ExecutionEngine._SLOT_HEARTBEAT_INTERVAL + 0.01 +SLEEP_SECONDS = 0.1 ITEM_A = {"id": "a"} ITEM_B = {"id": "b"} @@ -138,7 +132,7 @@ class MainTestCase(TestCase): @deferred_f_from_coro_f async def test_asyncio_delayed(self): async def start(spider): - await sleep(ASYNC_GEN_ERROR_MINIMUM_SECONDS) + await sleep(SLEEP_SECONDS) yield ITEM_A await self._test_start(start, [ITEM_A]) @@ -146,9 +140,7 @@ class MainTestCase(TestCase): @deferred_f_from_coro_f async def test_twisted_delayed(self): async def start(spider): - await maybe_deferred_to_future( - twisted_sleep(ASYNC_GEN_ERROR_MINIMUM_SECONDS) - ) + await maybe_deferred_to_future(twisted_sleep(SLEEP_SECONDS)) yield ITEM_A await self._test_start(start, [ITEM_A]) diff --git a/tests/test_spidermiddleware.py b/tests/test_spidermiddleware.py index 9728d0d81..9cd7b44ca 100644 --- a/tests/test_spidermiddleware.py +++ b/tests/test_spidermiddleware.py @@ -7,7 +7,6 @@ from unittest import mock import pytest from testfixtures import LogCapture from twisted.internet import defer -from twisted.internet.defer import inlineCallbacks from twisted.python.failure import Failure from twisted.trial.unittest import TestCase @@ -16,7 +15,11 @@ from scrapy.exceptions import _InvalidOutput from scrapy.http import Request, Response from scrapy.spiders import Spider from scrapy.utils.asyncgen import collect_asyncgen -from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future +from scrapy.utils.defer import ( + deferred_f_from_coro_f, + deferred_from_coro, + maybe_deferred_to_future, +) from scrapy.utils.test import get_crawler @@ -201,7 +204,7 @@ class ProcessSpiderExceptionSimpleIterableMiddleware: yield {"foo": 3} -class ProcessSpiderExceptionAsyncIterableMiddleware: +class ProcessSpiderExceptionAsyncIteratorMiddleware: async def process_spider_exception(self, response, exception, spider): yield {"foo": 1} d = defer.Deferred() @@ -332,8 +335,7 @@ class TestProcessStartSimple(TestBaseAsyncSpiderMiddleware): ITEM_TYPE = (Request, dict) MW_SIMPLE = ProcessStartSimpleMiddleware - @inlineCallbacks - def _get_processed_start(self, *mw_classes): + async def _get_processed_start(self, *mw_classes): class TestSpider(Spider): name = "test" @@ -348,15 +350,14 @@ class TestProcessStartSimple(TestBaseAsyncSpiderMiddleware): ) self.spider = self.crawler._create_spider() self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler) - results = yield self.mwman.process_start(self.spider) - return results + return await maybe_deferred_to_future(self.mwman.process_start(self.spider)) - @inlineCallbacks - def test_simple(self): + @deferred_f_from_coro_f + async def test_simple(self): """Simple mw""" - start = yield self._get_processed_start(self.MW_SIMPLE) + start = await self._get_processed_start(self.MW_SIMPLE) assert isasyncgen(start) - start_list = yield deferred_from_coro(collect_asyncgen(start)) + start_list = await collect_asyncgen(start) assert len(start_list) == self.RESULT_COUNT assert isinstance(start_list[0], self.ITEM_TYPE) @@ -516,7 +517,7 @@ class TestProcessSpiderException(TestBaseAsyncSpiderMiddleware): MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware MW_EXC_SIMPLE = ProcessSpiderExceptionSimpleIterableMiddleware - MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIterableMiddleware + MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIteratorMiddleware def _scrape_func(self, *args, **kwargs): 1 / 0 diff --git a/tests/test_spidermiddleware_process_start.py b/tests/test_spidermiddleware_process_start.py index 1913b4251..e2669a716 100644 --- a/tests/test_spidermiddleware_process_start.py +++ b/tests/test_spidermiddleware_process_start.py @@ -9,7 +9,7 @@ from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future from scrapy.utils.test import get_crawler -from .test_spider_start import ASYNC_GEN_ERROR_MINIMUM_SECONDS, twisted_sleep +from .test_spider_start import SLEEP_SECONDS, twisted_sleep ITEM_A = {"id": "a"} ITEM_B = {"id": "b"} @@ -19,7 +19,7 @@ ITEM_D = {"id": "d"} class AsyncioSleepSpiderMiddleware: async def process_start(self, start): - await sleep(ASYNC_GEN_ERROR_MINIMUM_SECONDS) + await sleep(SLEEP_SECONDS) async for item_or_request in start: yield item_or_request @@ -32,7 +32,7 @@ class NoOpSpiderMiddleware: class TwistedSleepSpiderMiddleware: async def process_start(self, start): - await maybe_deferred_to_future(twisted_sleep(ASYNC_GEN_ERROR_MINIMUM_SECONDS)) + await maybe_deferred_to_future(twisted_sleep(SLEEP_SECONDS)) async for item_or_request in start: yield item_or_request