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

Merge pull request #4978 from wRAR/asyncio-parse-asyncgen-proper-rebased 

Support for async callbacks
This commit is contained in:
Mikhail Korobov 2022-07-30 00:34:37 +05:00 committed by GitHub
parent 67011cd957 c7b90c6e1e
commit 52d93490f5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
27 changed files with 1405 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
docs/index.rst
View File

@ -225,10 +225,11 @@ Extending Scrapy
topics/downloader-middleware
topics/spider-middleware
topics/extensions
topics/api
topics/signals
topics/scheduler
topics/exporters
topics/components
topics/api
:doc:`topics/architecture`
@ -243,9 +244,6 @@ Extending Scrapy
:doc:`topics/extensions`
Extend Scrapy with your custom functionality
:doc:`topics/api`
Use it on extensions and middlewares to extend Scrapy functionality
:doc:`topics/signals`
See all available signals and how to work with them.
@ -255,6 +253,13 @@ Extending Scrapy
:doc:`topics/exporters`
Quickly export your scraped items to a file (XML, CSV, etc).
:doc:`topics/components`
Learn the common API and some good practices when building custom Scrapy
components.
:doc:`topics/api`
Use it on extensions and middlewares to extend Scrapy functionality.
All the rest
============

24
docs/topics/asyncio.rst
View File

@ -96,3 +96,27 @@ Futures. Scrapy provides two helpers for this:
down to Scrapy 2.0 (earlier versions do not support
:mod:`asyncio`), you can copy the implementation of these functions
into your own code.
.. _enforce-asyncio-requirement:
Enforcing asyncio as a requirement
==================================
If you are writing a :ref:`component <topics-components>` that requires asyncio
to work, use :func:`scrapy.utils.reactor.is_asyncio_reactor_installed` to
:ref:`enforce it as a requirement <enforce-component-requirements>`. For
example::
from scrapy.utils.reactor import is_asyncio_reactor_installed
class MyComponent:
def __init__(self):
if not is_asyncio_reactor_installed():
raise ValueError(
f"{MyComponent.__qualname__} requires the asyncio Twisted "
f"reactor. Make sure you have it configured in the "
f"TWISTED_REACTOR setting. See the asyncio documentation "
f"of Scrapy for more information."
)

84
docs/topics/components.rst Normal file
View File

@ -0,0 +1,84 @@
.. _topics-components:
==========
Components
==========
A Scrapy component is any class whose objects are created using
:func:`scrapy.utils.misc.create_instance`.
That includes the classes that you may assign to the following settings:
- :setting:`DNS_RESOLVER`
- :setting:`DOWNLOAD_HANDLERS`
- :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY`
- :setting:`DOWNLOADER_MIDDLEWARES`
- :setting:`DUPEFILTER_CLASS`
- :setting:`EXTENSIONS`
- :setting:`FEED_EXPORTERS`
- :setting:`FEED_STORAGES`
- :setting:`ITEM_PIPELINES`
- :setting:`SCHEDULER`
- :setting:`SCHEDULER_DISK_QUEUE`
- :setting:`SCHEDULER_MEMORY_QUEUE`
- :setting:`SCHEDULER_PRIORITY_QUEUE`
- :setting:`SPIDER_MIDDLEWARES`
Third-party Scrapy components may also let you define additional Scrapy
components, usually configurable through :ref:`settings <topics-settings>`, to
modify their behavior.
.. _enforce-component-requirements:
Enforcing component requirements
================================
Sometimes, your components may only be intended to work under certain
conditions. For example, they may require a minimum version of Scrapy to work as
intended, or they may require certain settings to have specific values.
In addition to describing those conditions in the documentation of your
component, it is a good practice to raise an exception from the ``__init__``
method of your component if those conditions are not met at run time.
In the case of :ref:`downloader middlewares <topics-downloader-middleware>`,
:ref:`extensions <topics-extensions>`, :ref:`item pipelines
<topics-item-pipeline>`, and :ref:`spider middlewares
<topics-spider-middleware>`, you should raise
:exc:`scrapy.exceptions.NotConfigured`, passing a description of the issue as a
parameter to the exception so that it is printed in the logs, for the user to
see. For other components, feel free to raise whatever other exception feels
right to you; for example, :exc:`RuntimeError` would make sense for a Scrapy
version mismatch, while :exc:`ValueError` may be better if the issue is the
value of a setting.
If your requirement is a minimum Scrapy version, you may use
:attr:`scrapy.__version__` to enforce your requirement. For example::
from pkg_resources import parse_version
import scrapy
class MyComponent:
def __init__(self):
if parse_version(scrapy.__version__) < parse_version('VERSION'):
raise RuntimeError(
f"{MyComponent.__qualname__} requires Scrapy VERSION or "
f"later, which allow defining the process_spider_output "
f"method of spider middlewares as an asynchronous "
f"generator."
)

133
docs/topics/coroutines.rst
View File

@ -19,14 +19,12 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
- :class:`~scrapy.Request` callbacks.
.. note:: The callback output is not processed until the whole callback
finishes.
If you are using any custom or third-party :ref:`spider middleware
<topics-spider-middleware>`, see :ref:`sync-async-spider-middleware`.
As a side effect, if the callback raises an exception, none of its
output is processed.
This is a known caveat of the current implementation that we aim to
address in a future version of Scrapy.
.. versionchanged:: VERSION
Output of async callbacks is now processed asynchronously instead of
collecting all of it first.
- The :meth:`process_item` method of
:ref:`item pipelines <topics-item-pipeline>`.
@ -41,12 +39,26 @@ hence use coroutine syntax (e.g. ``await``, ``async for``, ``async with``):
- :ref:`Signal handlers that support deferreds <signal-deferred>`.
Usage
=====
- The
:meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output`
method of :ref:`spider middlewares <topics-spider-middleware>`.
There are several use cases for coroutines in Scrapy. Code that would
return Deferreds when written for previous Scrapy versions, such as downloader
middlewares and signal handlers, can be rewritten to be shorter and cleaner::
It must be defined as 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`.
.. versionadded:: VERSION
General usage
=============
There are several use cases for coroutines in Scrapy.
Code that would return Deferreds when written for previous Scrapy versions,
such as downloader middlewares and signal handlers, can be rewritten to be
shorter and cleaner::
from itemadapter import ItemAdapter
@ -106,7 +118,100 @@ Common use cases for asynchronous code include:
* storing data in databases (in pipelines and middlewares);
* delaying the spider initialization until some external event (in the
:signal:`spider_opened` handler);
* calling asynchronous Scrapy methods like ``ExecutionEngine.download`` (see
:ref:`the screenshot pipeline example<ScreenshotPipeline>`).
* calling asynchronous Scrapy methods like :meth:`ExecutionEngine.download`
(see :ref:`the screenshot pipeline example<ScreenshotPipeline>`).
.. _aio-libs: https://github.com/aio-libs
.. _sync-async-spider-middleware:
Mixing synchronous and asynchronous spider middlewares
======================================================
.. versionadded:: VERSION
The output of a :class:`~scrapy.Request` callback is passed as the ``result``
parameter to the
:meth:`~scrapy.spidermiddlewares.SpiderMiddleware.process_spider_output` method
of the first :ref:`spider middleware <topics-spider-middleware>` from the
:ref:`list of active spider middlewares <topics-spider-middleware-setting>`.
Then the output of that ``process_spider_output`` method is passed to the
``process_spider_output`` method of the next spider middleware, and so on for
every active spider middleware.
Scrapy supports mixing :ref:`coroutine methods <async>` and synchronous methods
in this chain of calls.
However, if any of the ``process_spider_output`` methods is defined as a
synchronous method, and the previous ``Request`` callback or
``process_spider_output`` method is a coroutine, there are some drawbacks to
the asynchronous-to-synchronous conversion that Scrapy does so that the
synchronous ``process_spider_output`` method gets a synchronous iterable as its
``result`` parameter:
- The whole output of the previous ``Request`` callback or
``process_spider_output`` method is awaited at this point.
- If an exception raises while awaiting the output of the previous
``Request`` callback or ``process_spider_output`` method, none of that
output will be processed.
This contrasts with the regular behavior, where all items yielded before
an exception raises are processed.
Asynchronous-to-synchronous conversions are supported for backward
compatibility, but they are deprecated and will stop working in a future
version of Scrapy.
To avoid asynchronous-to-synchronous conversions, when defining ``Request``
callbacks as coroutine methods or when using spider middlewares whose
``process_spider_output`` method is an :term:`asynchronous generator`, all
active spider middlewares must either have their ``process_spider_output``
method defined as an asynchronous generator or :ref:`define a
process_spider_output_async method <universal-spider-middleware>`.
.. note:: When using third-party spider middlewares that only define a
synchronous ``process_spider_output`` method, consider
:ref:`making them universal <universal-spider-middleware>` through
:ref:`subclassing <tut-inheritance>`.
.. _universal-spider-middleware:
Universal spider middlewares
============================
.. versionadded:: VERSION
To allow writing a spider middleware that supports asynchronous execution of
its ``process_spider_output`` method in Scrapy VERSION and later (avoiding
:ref:`asynchronous-to-synchronous conversions <sync-async-spider-middleware>`)
while maintaining support for older Scrapy versions, you may define
``process_spider_output`` as a synchronous method and define an
:term:`asynchronous generator` version of that method with an alternative name:
``process_spider_output_async``.
For example::
class UniversalSpiderMiddleware:
def process_spider_output(self, response, result, spider):
for r in result:
# ... do something with r
yield r
async def process_spider_output_async(self, response, result, spider):
async for r in result:
# ... do something with r
yield r
.. note:: This is an interim measure to allow, for a time, to write code that
works in Scrapy VERSION and later without requiring
asynchronous-to-synchronous conversions, and works in earlier Scrapy
versions as well.
In some future version of Scrapy, however, this feature will be
deprecated and, eventually, in a later version of Scrapy, this
feature will be removed, and all spider middlewares will be expected
to define their ``process_spider_output`` method as an asynchronous
generator.

26
docs/topics/spider-middleware.rst
View File

@ -102,27 +102,47 @@ object gives you access, for example, to the :ref:`settings <topics-settings>`.
it has processed the response.
:meth:`process_spider_output` must return an iterable of
:class:`~scrapy.Request` objects and :ref:`item object
:class:`~scrapy.Request` objects and :ref:`item objects
<topics-items>`.
.. versionchanged:: VERSION
This method may be defined as an :term:`asynchronous generator`, in
which case ``result`` is an :term:`asynchronous iterable`.
Consider defining this method as an :term:`asynchronous generator`,
which will be a requirement in a future version of Scrapy. However, if
you plan on sharing your spider middleware with other people, consider
either :ref:`enforcing Scrapy VERSION <enforce-component-requirements>`
as a minimum requirement of your spider middleware, or :ref:`making
your spider middleware universal <universal-spider-middleware>` so that
it works with Scrapy versions earlier than Scrapy VERSION.
:param response: the response which generated this output from the
spider
:type response: :class:`~scrapy.http.Response` object
:param result: the result returned by the spider
:type result: an iterable of :class:`~scrapy.Request` objects and
:ref:`item object <topics-items>`
:ref:`item objects <topics-items>`
:param spider: the spider whose result is being processed
:type spider: :class:`~scrapy.Spider` object
.. method:: process_spider_output_async(response, result, spider)
.. versionadded:: VERSION
If defined, this method must be an :term:`asynchronous generator`,
which will be called instead of :meth:`process_spider_output` if
``result`` is an :term:`asynchronous iterable`.
.. method:: process_spider_exception(response, exception, spider)
This method is called when a spider or :meth:`process_spider_output`
method (from a previous spider middleware) raises an exception.
:meth:`process_spider_exception` should return either ``None`` or an
iterable of :class:`~scrapy.Request` or :ref:`item <topics-items>`
iterable of :class:`~scrapy.Request` or :ref:`item <topics-items>`
objects.
If it returns ``None``, Scrapy will continue processing this exception,

1
pylintrc
View File

@ -47,6 +47,7 @@ disable=abstract-method,
keyword-arg-before-vararg,
line-too-long,
logging-format-interpolation,
logging-fstring-interpolation,
logging-not-lazy,
lost-exception,
method-hidden,

28
scrapy/core/scraper.py
View File

@ -1,9 +1,8 @@
"""This module implements the Scraper component which parses responses and
extracts information from them"""
import logging
from collections import deque
from typing import Any, Deque, Iterable, Optional, Set, Tuple, Union
from typing import Any, AsyncGenerator, AsyncIterable, Deque, Generator, Iterable, Optional, Set, Tuple, Union
from itemadapter import is_item
from twisted.internet.defer import Deferred, inlineCallbacks
@ -13,7 +12,15 @@ from scrapy import signals, Spider
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,
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
@ -185,12 +192,19 @@ class Scraper:
spider=spider
)
def handle_spider_output(self, result: Iterable, request: Request, response: Response, spider: Spider) -> Deferred:
def handle_spider_output(self, result: Union[Iterable, AsyncIterable], request: Request,
response: Response, spider: Spider) -> Deferred:
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)
it: Union[Generator, AsyncGenerator]
if isinstance(result, AsyncIterable):
it = aiter_errback(result, self.handle_spider_error, request, response, spider)
dfd = 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: Any, request: Request, response: Response,

196
scrapy/core/spidermw.py
View File

@ -3,32 +3,42 @@ Spider Middleware manager
See documentation in docs/topics/spider-middleware.rst
"""
import logging
from inspect import isasyncgenfunction, iscoroutine
from itertools import islice
from typing import Any, Callable, Generator, Iterable, Union, cast
from typing import Any, AsyncGenerator, AsyncIterable, Callable, Generator, Iterable, Tuple, Union, cast
from twisted.internet.defer import Deferred
from twisted.internet.defer import Deferred, inlineCallbacks
from twisted.python.failure import Failure
from scrapy import Request, Spider
from scrapy.exceptions import _InvalidOutput
from scrapy.http import Response
from scrapy.middleware import MiddlewareManager
from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
from scrapy.utils.conf import build_component_list
from scrapy.utils.defer import mustbe_deferred
from scrapy.utils.python import MutableChain
from scrapy.utils.defer import mustbe_deferred, deferred_from_coro, deferred_f_from_coro_f, maybe_deferred_to_future
from scrapy.utils.python import MutableAsyncChain, MutableChain
logger = logging.getLogger(__name__)
ScrapeFunc = Callable[[Union[Response, Failure], Request, Spider], Any]
def _isiterable(o) -> bool:
return isinstance(o, Iterable)
return isinstance(o, (Iterable, AsyncIterable))
class SpiderMiddlewareManager(MiddlewareManager):
component_name = 'spider middleware'
def __init__(self, *middlewares):
super().__init__(*middlewares)
self.downgrade_warning_done = False
@classmethod
def _get_mwlist_from_settings(cls, settings):
return build_component_list(settings.getwithbase('SPIDER_MIDDLEWARES'))
@ -39,7 +49,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
self.methods['process_spider_input'].append(mw.process_spider_input)
if hasattr(mw, 'process_start_requests'):
self.methods['process_start_requests'].appendleft(mw.process_start_requests)
process_spider_output = getattr(mw, 'process_spider_output', None)
process_spider_output = self._get_async_method_pair(mw, 'process_spider_output')
self.methods['process_spider_output'].appendleft(process_spider_output)
process_spider_exception = getattr(mw, 'process_spider_exception', None)
self.methods['process_spider_exception'].appendleft(process_spider_exception)
@ -51,7 +61,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
try:
result = method(response=response, spider=spider)
if result is not None:
msg = (f"Middleware {method.__qualname__} must return None "
msg = (f"{method.__qualname__} must return None "
f"or raise an exception, got {type(result)}")
raise _InvalidOutput(msg)
except _InvalidOutput:
@ -60,17 +70,35 @@ class SpiderMiddlewareManager(MiddlewareManager):
return scrape_func(Failure(), request, spider)
return scrape_func(response, request, spider)
def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Iterable,
exception_processor_index: int, recover_to: MutableChain) -> Generator:
try:
for r in iterable:
yield r
except Exception as ex:
exception_result = self._process_spider_exception(response, spider, Failure(ex),
exception_processor_index)
if isinstance(exception_result, Failure):
raise
recover_to.extend(exception_result)
def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Union[Iterable, AsyncIterable],
exception_processor_index: int, recover_to: Union[MutableChain, MutableAsyncChain]
) -> Union[Generator, AsyncGenerator]:
def process_sync(iterable: Iterable):
try:
for r in iterable:
yield r
except Exception as ex:
exception_result = self._process_spider_exception(response, spider, Failure(ex),
exception_processor_index)
if isinstance(exception_result, Failure):
raise
recover_to.extend(exception_result)
async def process_async(iterable: AsyncIterable):
try:
async for r in iterable:
yield r
except Exception as ex:
exception_result = self._process_spider_exception(response, spider, Failure(ex),
exception_processor_index)
if isinstance(exception_result, Failure):
raise
recover_to.extend(exception_result)
if isinstance(iterable, AsyncIterable):
return process_async(iterable)
return process_sync(iterable)
def _process_spider_exception(self, response: Response, spider: Spider, _failure: Failure,
start_index: int = 0) -> Union[Failure, MutableChain]:
@ -82,30 +110,83 @@ class SpiderMiddlewareManager(MiddlewareManager):
for method_index, method in enumerate(method_list, start=start_index):
if method is None:
continue
method = cast(Callable, method)
result = method(response=response, exception=exception, spider=spider)
if _isiterable(result):
# stop exception handling by handing control over to the
# process_spider_output chain if an iterable has been returned
return self._process_spider_output(response, spider, result, method_index + 1)
dfd: Deferred = self._process_spider_output(response, spider, result, method_index + 1)
# _process_spider_output() returns a Deferred only because of downgrading so this can be
# simplified when downgrading is removed.
if dfd.called:
# the result is available immediately if _process_spider_output didn't do downgrading
return dfd.result
else:
# we forbid waiting here because otherwise we would need to return a deferred from
# _process_spider_exception too, which complicates the architecture
msg = f"Async iterable returned from {method.__qualname__} cannot be downgraded"
raise _InvalidOutput(msg)
elif result is None:
continue
else:
msg = (f"Middleware {method.__qualname__} must return None "
msg = (f"{method.__qualname__} must return None "
f"or an iterable, got {type(result)}")
raise _InvalidOutput(msg)
return _failure
# This method cannot be made async def, as _process_spider_exception relies on the Deferred result
# being available immediately which doesn't work when it's a wrapped coroutine.
# It also needs @inlineCallbacks only because of downgrading so it can be removed when downgrading is removed.
@inlineCallbacks
def _process_spider_output(self, response: Response, spider: Spider,
result: Iterable, start_index: int = 0) -> MutableChain:
result: Union[Iterable, AsyncIterable], start_index: int = 0
) -> Deferred:
# 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()
recovered: Union[MutableChain, MutableAsyncChain]
last_result_is_async = isinstance(result, AsyncIterable)
if last_result_is_async:
recovered = MutableAsyncChain()
else:
recovered = MutableChain()
# There are three cases for the middleware: def foo, async def foo, def foo + async def foo_async.
# 1. def foo. Sync iterables are passed as is, async ones are downgraded.
# 2. async def foo. Sync iterables are upgraded, async ones are passed as is.
# 3. def foo + async def foo_async. Iterables are passed to the respective method.
# Storing methods and method tuples in the same list is weird but we should be able to roll this back
# when we drop this compatibility feature.
method_list = islice(self.methods['process_spider_output'], start_index, None)
for method_index, method in enumerate(method_list, start=start_index):
if method is None:
for method_index, method_pair in enumerate(method_list, start=start_index):
if method_pair is None:
continue
need_upgrade = need_downgrade = False
if isinstance(method_pair, tuple):
# This tuple handling is only needed until _async compatibility methods are removed.
method_sync, method_async = method_pair
method = method_async if last_result_is_async else method_sync
else:
method = method_pair
if not last_result_is_async and isasyncgenfunction(method):
need_upgrade = True
elif last_result_is_async and not isasyncgenfunction(method):
need_downgrade = True
try:
if need_upgrade:
# Iterable -> AsyncIterable
result = as_async_generator(result)
elif need_downgrade:
if not self.downgrade_warning_done:
logger.warning(f"Async iterable passed to {method.__qualname__} "
f"was downgraded to a non-async one")
self.downgrade_warning_done = True
assert isinstance(result, AsyncIterable)
# AsyncIterable -> Iterable
result = yield deferred_from_coro(collect_asyncgen(result))
if isinstance(recovered, AsyncIterable):
recovered_collected = yield deferred_from_coro(collect_asyncgen(recovered))
recovered = MutableChain(recovered_collected)
# might fail directly if the output value is not a generator
result = method(response=response, result=result, spider=spider)
except Exception as ex:
@ -116,28 +197,77 @@ class SpiderMiddlewareManager(MiddlewareManager):
if _isiterable(result):
result = self._evaluate_iterable(response, spider, result, method_index + 1, recovered)
else:
msg = (f"Middleware {method.__qualname__} must return an "
f"iterable, got {type(result)}")
if iscoroutine(result):
result.close() # Silence warning about not awaiting
msg = (
f"{method.__qualname__} must be an asynchronous "
f"generator (i.e. use yield)"
)
else:
msg = (
f"{method.__qualname__} must return an iterable, got "
f"{type(result)}"
)
raise _InvalidOutput(msg)
last_result_is_async = isinstance(result, AsyncIterable)
return MutableChain(result, recovered)
if last_result_is_async:
return MutableAsyncChain(result, recovered)
else:
return MutableChain(result, recovered) # type: ignore[arg-type]
def _process_callback_output(self, response: Response, spider: Spider, result: Iterable) -> MutableChain:
recovered = MutableChain()
async def _process_callback_output(self, response: Response, spider: Spider, result: Union[Iterable, AsyncIterable]
) -> Union[MutableChain, MutableAsyncChain]:
recovered: Union[MutableChain, MutableAsyncChain]
if isinstance(result, AsyncIterable):
recovered = MutableAsyncChain()
else:
recovered = MutableChain()
result = self._evaluate_iterable(response, spider, result, 0, recovered)
return MutableChain(self._process_spider_output(response, spider, result), recovered)
result = await maybe_deferred_to_future(self._process_spider_output(response, spider, result))
if isinstance(result, AsyncIterable):
return MutableAsyncChain(result, recovered)
else:
if isinstance(recovered, AsyncIterable):
recovered_collected = await collect_asyncgen(recovered)
recovered = MutableChain(recovered_collected)
return MutableChain(result, recovered) # type: ignore[arg-type]
def scrape_response(self, scrape_func: ScrapeFunc, response: Response, request: Request,
spider: Spider) -> Deferred:
def process_callback_output(result: Iterable) -> MutableChain:
return self._process_callback_output(response, spider, result)
async def process_callback_output(result: Union[Iterable, AsyncIterable]
) -> Union[MutableChain, MutableAsyncChain]:
return await self._process_callback_output(response, spider, result)
def process_spider_exception(_failure: Failure) -> Union[Failure, MutableChain]:
return self._process_spider_exception(response, spider, _failure)
dfd = mustbe_deferred(self._process_spider_input, scrape_func, response, request, spider)
dfd.addCallbacks(callback=process_callback_output, errback=process_spider_exception)
dfd.addCallbacks(callback=deferred_f_from_coro_f(process_callback_output), errback=process_spider_exception)
return dfd
def process_start_requests(self, start_requests, spider: Spider) -> Deferred:
return self._process_chain('process_start_requests', start_requests, spider)
# This method is only needed until _async compatibility methods are removed.
@staticmethod
def _get_async_method_pair(mw: Any, methodname: str) -> Union[None, Callable, Tuple[Callable, Callable]]:
normal_method = getattr(mw, methodname, None)
methodname_async = methodname + "_async"
async_method = getattr(mw, methodname_async, None)
if not async_method:
return normal_method
if not normal_method:
logger.error(f"Middleware {mw.__qualname__} has {methodname_async} "
f"without {methodname}, skipping this method.")
return None
if not isasyncgenfunction(async_method):
logger.error(f"{async_method.__qualname__} is not "
f"an async generator function, skipping this method.")
return normal_method
if isasyncgenfunction(normal_method):
logger.error(f"{normal_method.__qualname__} is an async "
f"generator function while {methodname_async} exists, "
f"skipping both methods.")
return None
return normal_method, async_method

7
scrapy/middleware.py
View File

@ -1,7 +1,7 @@
import logging
import pprint
from collections import defaultdict, deque
from typing import Callable, Deque, Dict, Optional, cast, Iterable
from typing import Callable, Deque, Dict, Iterable, Tuple, Union, cast
from twisted.internet.defer import Deferred
@ -21,8 +21,9 @@ class MiddlewareManager:
def __init__(self, *middlewares):
self.middlewares = middlewares
# Optional because process_spider_output and process_spider_exception can be None
self.methods: Dict[str, Deque[Optional[Callable]]] = defaultdict(deque)
# Only process_spider_output and process_spider_exception can be None.
# Only process_spider_output can be a tuple, and only until _async compatibility methods are removed.
self.methods: Dict[str, Deque[Union[None, Callable, Tuple[Callable, Callable]]]] = defaultdict(deque)
for mw in middlewares:
self._add_middleware(mw)

50
scrapy/spidermiddlewares/depth.py
View File

@ -28,31 +28,39 @@ class DepthMiddleware:
return cls(maxdepth, crawler.stats, verbose, prio)
def process_spider_output(self, response, result, spider):
def _filter(request):
if isinstance(request, Request):
depth = response.meta['depth'] + 1
request.meta['depth'] = depth
if self.prio:
request.priority -= depth * self.prio
if self.maxdepth and depth > self.maxdepth:
logger.debug(
"Ignoring link (depth > %(maxdepth)d): %(requrl)s ",
{'maxdepth': self.maxdepth, 'requrl': request.url},
extra={'spider': spider}
)
return False
else:
if self.verbose_stats:
self.stats.inc_value(f'request_depth_count/{depth}',
spider=spider)
self.stats.max_value('request_depth_max', depth,
spider=spider)
return True
self._init_depth(response, spider)
return (r for r in result or () if self._filter(r, response, spider))
async def process_spider_output_async(self, response, result, spider):
self._init_depth(response, spider)
async for r in result or ():
if self._filter(r, response, spider):
yield r
def _init_depth(self, response, spider):
# 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))
def _filter(self, request, response, spider):
if not isinstance(request, Request):
return True
depth = response.meta['depth'] + 1
request.meta['depth'] = depth
if self.prio:
request.priority -= depth * self.prio
if self.maxdepth and depth > self.maxdepth:
logger.debug(
"Ignoring link (depth > %(maxdepth)d): %(requrl)s ",
{'maxdepth': self.maxdepth, 'requrl': request.url},
extra={'spider': spider}
)
return False
if self.verbose_stats:
self.stats.inc_value(f'request_depth_count/{depth}',
spider=spider)
self.stats.max_value('request_depth_max', depth,
spider=spider)
return True

36
scrapy/spidermiddlewares/offsite.py
View File

@ -26,21 +26,27 @@ 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
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
return (r for r in result or () if self._filter(r, spider))
async def process_spider_output_async(self, response, result, spider):
async for r in result or ():
if self._filter(r, spider):
yield r
def _filter(self, request, spider) -> bool:
if not isinstance(request, Request):
return True
if request.dont_filter or self.should_follow(request, spider):
return True
domain = urlparse_cached(request).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': request}, extra={'spider': spider})
self.stats.inc_value('offsite/domains', spider=spider)
self.stats.inc_value('offsite/filtered', spider=spider)
return False
def should_follow(self, request, spider):
regex = self.host_regex

19
scrapy/spidermiddlewares/referer.py
View File

@ -333,13 +333,18 @@ class RefererMiddleware:
return cls() if cls else self.default_policy()
def process_spider_output(self, response, result, spider):
def _set_referer(r):
if isinstance(r, Request):
referrer = self.policy(response, r).referrer(response.url, r.url)
if referrer is not None:
r.headers.setdefault('Referer', referrer)
return r
return (_set_referer(r) for r in result or ())
return (self._set_referer(r, response) for r in result or ())
async def process_spider_output_async(self, response, result, spider):
async for r in result or ():
yield self._set_referer(r, response)
def _set_referer(self, r, response):
if isinstance(r, Request):
referrer = self.policy(response, r).referrer(response.url, r.url)
if referrer is not None:
r.headers.setdefault('Referer', referrer)
return r
def request_scheduled(self, request, spider):
# check redirected request to patch "Referer" header if necessary

28
scrapy/spidermiddlewares/urllength.py
View File

@ -25,16 +25,20 @@ class UrlLengthMiddleware:
return cls(maxlength)
def process_spider_output(self, response, result, spider):
def _filter(request):
if isinstance(request, Request) and len(request.url) > self.maxlength:
logger.info(
"Ignoring link (url length > %(maxlength)d): %(url)s ",
{'maxlength': self.maxlength, 'url': request.url},
extra={'spider': spider}
)
spider.crawler.stats.inc_value('urllength/request_ignored_count', spider=spider)
return False
else:
return True
return (r for r in result or () if self._filter(r, spider))
return (r for r in result or () if _filter(r))
async def process_spider_output_async(self, response, result, spider):
async for r in result or ():
if self._filter(r, spider):
yield r
def _filter(self, request, spider):
if isinstance(request, Request) and len(request.url) > self.maxlength:
logger.info(
"Ignoring link (url length > %(maxlength)d): %(url)s ",
{'maxlength': self.maxlength, 'url': request.url},
extra={'spider': spider}
)
spider.crawler.stats.inc_value('urllength/request_ignored_count', spider=spider)
return False
return True

12
scrapy/utils/asyncgen.py
View File

@ -1,4 +1,4 @@
from collections.abc import AsyncIterable
from typing import AsyncGenerator, AsyncIterable, Iterable, Union
async def collect_asyncgen(result: AsyncIterable):
@ -6,3 +6,13 @@ async def collect_asyncgen(result: AsyncIterable):
async for x in result:
results.append(x)
return results
async def as_async_generator(it: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
""" Wraps an iterable (sync or async) into an async generator. """
if isinstance(it, AsyncIterable):
async for r in it:
yield r
else:
for r in it:
yield r

122
scrapy/utils/defer.py
View File

@ -7,10 +7,15 @@ from asyncio import Future
from functools import wraps
from typing import (
Any,
AsyncGenerator,
AsyncIterable,
Callable,
Coroutine,
Generator,
Iterable,
Iterator,
List,
Optional,
Union
)
@ -87,6 +92,109 @@ def parallel(iterable: Iterable, count: int, callable: Callable, *args, **named)
return DeferredList([coop.coiterate(work) for _ in range(count)])
class _AsyncCooperatorAdapter(Iterator):
""" A class that wraps an async iterable 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, aiterable: AsyncIterable, callable: Callable, *callable_args, **callable_kwargs):
self.aiterator = aiterable.__aiter__()
self.callable = callable
self.callable_args = callable_args
self.callable_kwargs = callable_kwargs
self.finished = False
self.waiting_deferreds: List[Deferred] = []
self.anext_deferred: Optional[Deferred] = None
def _callback(self, result: Any) -> None:
# 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 isinstance(result, Deferred):
result.chainDeferred(d)
else:
d.callback(None)
if self.waiting_deferreds:
self._call_anext()
def _errback(self, failure: Failure) -> None:
# 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:
d.callback(None)
def _call_anext(self) -> None:
# 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 __next__(self) -> Deferred:
# This puts a new Deferred into self.waiting_deferreds and returns it.
# It also calls __anext__() if needed.
if self.finished:
raise StopIteration
d = Deferred()
self.waiting_deferreds.append(d)
if not self.anext_deferred:
self._call_anext()
return d
def parallel_async(async_iterable: AsyncIterable, count: int, callable: Callable, *args, **named) -> DeferredList:
""" Like parallel but for async iterators """
coop = Cooperator()
work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named)
dl = DeferredList([coop.coiterate(work) for _ in range(count)])
return dl
def process_chain(callbacks: Iterable[Callable], input, *a, **kw) -> Deferred:
"""Return a Deferred built by chaining the given callbacks"""
d = Deferred()
@ -136,6 +244,20 @@ def iter_errback(iterable: Iterable, errback: Callable, *a, **kw) -> Generator:
errback(failure.Failure(), *a, **kw)
async def aiter_errback(aiterable: AsyncIterable, errback: Callable, *a, **kw) -> AsyncGenerator:
"""Wraps an async iterable calling an errback if an error is caught while
iterating it. Similar to scrapy.utils.defer.iter_errback()
"""
it = aiterable.__aiter__()
while True:
try:
yield await it.__anext__()
except StopAsyncIteration:
break
except Exception:
errback(failure.Failure(), *a, **kw)
def deferred_from_coro(o) -> Any:
"""Converts a coroutine into a Deferred, or returns the object as is if it isn't a coroutine"""
if isinstance(o, Deferred):

29
scrapy/utils/python.py
View File

@ -8,11 +8,12 @@ import re
import sys
import warnings
import weakref
from collections.abc import Iterable
from functools import partial, wraps
from itertools import chain
from typing import AsyncGenerator, AsyncIterable, Iterable, Union
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.asyncgen import as_async_generator
from scrapy.utils.decorators import deprecated
@ -344,7 +345,7 @@ class MutableChain(Iterable):
def __init__(self, *args: Iterable):
self.data = chain.from_iterable(args)
def extend(self, *iterables: Iterable):
def extend(self, *iterables: Iterable) -> None:
self.data = chain(self.data, chain.from_iterable(iterables))
def __iter__(self):
@ -356,3 +357,27 @@ class MutableChain(Iterable):
@deprecated("scrapy.utils.python.MutableChain.__next__")
def next(self):
return self.__next__()
async def _async_chain(*iterables: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
for it in iterables:
async for o in as_async_generator(it):
yield o
class MutableAsyncChain(AsyncIterable):
"""
Similar to MutableChain but for async iterables
"""
def __init__(self, *args: Union[Iterable, AsyncIterable]):
self.data = _async_chain(*args)
def extend(self, *iterables: Union[Iterable, AsyncIterable]) -> None:
self.data = _async_chain(self.data, _async_chain(*iterables))
def __aiter__(self):
return self
async def __anext__(self):
return await self.data.__anext__()

8
scrapy/utils/spider.py
View File

@ -4,7 +4,6 @@ import logging
from scrapy.spiders import Spider
from scrapy.utils.defer import deferred_from_coro
from scrapy.utils.misc import arg_to_iter
from scrapy.utils.asyncgen import collect_asyncgen
logger = logging.getLogger(__name__)
@ -12,14 +11,13 @@ logger = logging.getLogger(__name__)
def iterate_spider_output(result):
if inspect.isasyncgen(result):
d = deferred_from_coro(collect_asyncgen(result))
d.addCallback(iterate_spider_output)
return d
return result
elif inspect.iscoroutine(result):
d = deferred_from_coro(result)
d.addCallback(iterate_spider_output)
return d
return arg_to_iter(result)
else:
return arg_to_iter(deferred_from_coro(result))
def iter_spider_classes(module):

7
scrapy/utils/test.py
View File

@ -115,3 +115,10 @@ def mock_google_cloud_storage():
bucket_mock.blob.return_value = blob_mock
return (client_mock, bucket_mock, blob_mock)
def get_web_client_agent_req(url):
from twisted.internet import reactor
from twisted.web.client import Agent # imports twisted.internet.reactor
agent = Agent(reactor)
return agent.request(b'GET', url.encode('utf-8'))

38
tests/spiders.py
View File

@ -14,7 +14,8 @@ from scrapy.item import Item
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import Spider
from scrapy.spiders.crawl import CrawlSpider, Rule
from scrapy.utils.test import get_from_asyncio_queue
from scrapy.utils.defer import deferred_to_future, maybe_deferred_to_future
from scrapy.utils.test import get_from_asyncio_queue, get_web_client_agent_req
class MockServerSpider(Spider):
@ -148,6 +149,41 @@ class AsyncDefAsyncioReqsReturnSpider(SimpleSpider):
return reqs
class AsyncDefAsyncioGenExcSpider(SimpleSpider):
name = 'asyncdef_asyncio_gen_exc'
async def parse(self, response):
for i in range(10):
await asyncio.sleep(0.1)
yield {'foo': i}
if i > 5:
raise ValueError("Stopping the processing")
class AsyncDefDeferredDirectSpider(SimpleSpider):
name = 'asyncdef_deferred_direct'
async def parse(self, response):
resp = await get_web_client_agent_req(self.mockserver.url("/status?n=200"))
yield {'code': resp.code}
class AsyncDefDeferredWrappedSpider(SimpleSpider):
name = 'asyncdef_deferred_wrapped'
async def parse(self, response):
resp = await deferred_to_future(get_web_client_agent_req(self.mockserver.url("/status?n=200")))
yield {'code': resp.code}
class AsyncDefDeferredMaybeWrappedSpider(SimpleSpider):
name = 'asyncdef_deferred_wrapped'
async def parse(self, response):
resp = await maybe_deferred_to_future(get_web_client_agent_req(self.mockserver.url("/status?n=200")))
yield {'code': resp.code}
class AsyncDefAsyncioGenSpider(SimpleSpider):
name = 'asyncdef_asyncio_gen'

33
tests/test_crawl.py
View File

@ -23,12 +23,16 @@ from tests import NON_EXISTING_RESOLVABLE
from tests.mockserver import MockServer
from tests.spiders import (
AsyncDefAsyncioGenComplexSpider,
AsyncDefAsyncioGenExcSpider,
AsyncDefAsyncioGenLoopSpider,
AsyncDefAsyncioGenSpider,
AsyncDefAsyncioReqsReturnSpider,
AsyncDefAsyncioReturnSingleElementSpider,
AsyncDefAsyncioReturnSpider,
AsyncDefAsyncioSpider,
AsyncDefDeferredDirectSpider,
AsyncDefDeferredMaybeWrappedSpider,
AsyncDefDeferredWrappedSpider,
AsyncDefSpider,
BrokenStartRequestsSpider,
BytesReceivedCallbackSpider,
@ -458,6 +462,18 @@ class CrawlSpiderTestCase(TestCase):
for i in range(10):
self.assertIn({'foo': i}, items)
@mark.only_asyncio()
@defer.inlineCallbacks
def test_async_def_asyncgen_parse_exc(self):
log, items, stats = yield self._run_spider(AsyncDefAsyncioGenExcSpider)
log = str(log)
self.assertIn("Spider error processing", log)
self.assertIn("ValueError", log)
itemcount = stats.get_value('item_scraped_count')
self.assertEqual(itemcount, 7)
for i in range(7):
self.assertIn({'foo': i}, items)
@mark.only_asyncio()
@defer.inlineCallbacks
def test_async_def_asyncgen_parse_complex(self):
@ -477,6 +493,23 @@ class CrawlSpiderTestCase(TestCase):
for req_id in range(3):
self.assertIn(f"Got response 200, req_id {req_id}", str(log))
@mark.only_not_asyncio()
@defer.inlineCallbacks
def test_async_def_deferred_direct(self):
_, items, _ = yield self._run_spider(AsyncDefDeferredDirectSpider)
self.assertEqual(items, [{'code': 200}])
@mark.only_asyncio()
@defer.inlineCallbacks
def test_async_def_deferred_wrapped(self):
log, items, _ = yield self._run_spider(AsyncDefDeferredWrappedSpider)
self.assertEqual(items, [{'code': 200}])
@defer.inlineCallbacks
def test_async_def_deferred_maybe_wrapped(self):
_, items, _ = yield self._run_spider(AsyncDefDeferredMaybeWrappedSpider)
self.assertEqual(items, [{'code': 200}])
@defer.inlineCallbacks
def test_response_ssl_certificate_none(self):
crawler = get_crawler(SingleRequestSpider)

430
tests/test_spidermiddleware.py
View File

@ -1,11 +1,17 @@
import collections.abc
from typing import Optional
from unittest import mock
from testfixtures import LogCapture
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.asyncgen import collect_asyncgen
from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
from scrapy.utils.test import get_crawler
from scrapy.core.spidermw import SpiderMiddlewareManager
@ -101,3 +107,427 @@ class ProcessSpiderExceptionReRaise(SpiderMiddlewareTestCase):
result = self._scrape_response()
self.assertIsInstance(result, Failure)
self.assertIsInstance(result.value, ZeroDivisionError)
class BaseAsyncSpiderMiddlewareTestCase(SpiderMiddlewareTestCase):
""" Helpers for testing sync, async and mixed middlewares.
Should work for process_spider_output and, when it's supported, process_start_requests.
"""
RESULT_COUNT = 3 # to simplify checks, let everything return 3 objects
@staticmethod
def _construct_mw_setting(*mw_classes, start_index: Optional[int] = None):
if start_index is None:
start_index = 10
return {i: c for c, i in enumerate(mw_classes, start=start_index)}
def _scrape_func(self, *args, **kwargs):
yield {'foo': 1}
yield {'foo': 2}
yield {'foo': 3}
@defer.inlineCallbacks
def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES_BASE': {}, 'SPIDER_MIDDLEWARES': setting})
self.spider = self.crawler._create_spider('foo')
self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
result = yield self.mwman.scrape_response(self._scrape_func, self.response, self.request, self.spider)
return result
@defer.inlineCallbacks
def _test_simple_base(self, *mw_classes, downgrade: bool = False, start_index: Optional[int] = None):
with LogCapture() as log:
result = yield self._get_middleware_result(*mw_classes, start_index=start_index)
self.assertIsInstance(result, collections.abc.Iterable)
result_list = list(result)
self.assertEqual(len(result_list), self.RESULT_COUNT)
self.assertIsInstance(result_list[0], self.ITEM_TYPE)
self.assertEqual("downgraded to a non-async" in str(log), downgrade)
@defer.inlineCallbacks
def _test_asyncgen_base(self, *mw_classes, downgrade: bool = False, start_index: Optional[int] = None):
with LogCapture() as log:
result = yield self._get_middleware_result(*mw_classes, start_index=start_index)
self.assertIsInstance(result, collections.abc.AsyncIterator)
result_list = yield deferred_from_coro(collect_asyncgen(result))
self.assertEqual(len(result_list), self.RESULT_COUNT)
self.assertIsInstance(result_list[0], self.ITEM_TYPE)
self.assertEqual("downgraded to a non-async" in str(log), downgrade)
class ProcessSpiderOutputSimpleMiddleware:
def process_spider_output(self, response, result, spider):
for r in result:
yield r
class ProcessSpiderOutputAsyncGenMiddleware:
async def process_spider_output(self, response, result, spider):
async for r in result:
yield r
class ProcessSpiderOutputUniversalMiddleware:
def process_spider_output(self, response, result, spider):
for r in result:
yield r
async def process_spider_output_async(self, response, result, spider):
async for r in result:
yield r
class ProcessSpiderExceptionSimpleIterableMiddleware:
def process_spider_exception(self, response, exception, spider):
yield {'foo': 1}
yield {'foo': 2}
yield {'foo': 3}
class ProcessSpiderExceptionAsyncIterableMiddleware:
async def process_spider_exception(self, response, exception, spider):
yield {'foo': 1}
d = defer.Deferred()
from twisted.internet import reactor
reactor.callLater(0, d.callback, None)
await maybe_deferred_to_future(d)
yield {'foo': 2}
yield {'foo': 3}
class ProcessSpiderOutputSimple(BaseAsyncSpiderMiddlewareTestCase):
""" process_spider_output tests for simple callbacks"""
ITEM_TYPE = dict
MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
def test_simple(self):
""" Simple mw """
return self._test_simple_base(self.MW_SIMPLE)
def test_asyncgen(self):
""" Asyncgen mw; upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN)
def test_simple_asyncgen(self):
""" Simple mw -> asyncgen mw; upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN,
self.MW_SIMPLE)
def test_asyncgen_simple(self):
""" Asyncgen mw -> simple mw; upgrade then downgrade """
return self._test_simple_base(self.MW_SIMPLE,
self.MW_ASYNCGEN,
downgrade=True)
def test_universal(self):
""" Universal mw """
return self._test_simple_base(self.MW_UNIVERSAL)
def test_universal_simple(self):
""" Universal mw -> simple mw """
return self._test_simple_base(self.MW_SIMPLE,
self.MW_UNIVERSAL)
def test_simple_universal(self):
""" Simple mw -> universal mw """
return self._test_simple_base(self.MW_UNIVERSAL,
self.MW_SIMPLE)
def test_universal_asyncgen(self):
""" Universal mw -> asyncgen mw; upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN,
self.MW_UNIVERSAL)
def test_asyncgen_universal(self):
""" Asyncgen mw -> universal mw; upgrade """
return self._test_asyncgen_base(self.MW_UNIVERSAL,
self.MW_ASYNCGEN)
class ProcessSpiderOutputAsyncGen(ProcessSpiderOutputSimple):
""" process_spider_output tests for async generator callbacks """
async def _scrape_func(self, *args, **kwargs):
for item in super()._scrape_func():
yield item
def test_simple(self):
""" Simple mw; downgrade """
return self._test_simple_base(self.MW_SIMPLE,
downgrade=True)
def test_simple_asyncgen(self):
""" Simple mw -> asyncgen mw; downgrade then upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN,
self.MW_SIMPLE,
downgrade=True)
def test_universal(self):
""" Universal mw """
return self._test_asyncgen_base(self.MW_UNIVERSAL)
def test_universal_simple(self):
""" Universal mw -> simple mw; downgrade """
return self._test_simple_base(self.MW_SIMPLE,
self.MW_UNIVERSAL,
downgrade=True)
def test_simple_universal(self):
""" Simple mw -> universal mw; downgrade """
return self._test_simple_base(self.MW_UNIVERSAL,
self.MW_SIMPLE,
downgrade=True)
class ProcessSpiderOutputNonIterableMiddleware:
def process_spider_output(self, response, result, spider):
return
class ProcessSpiderOutputCoroutineMiddleware:
async def process_spider_output(self, response, result, spider):
results = []
for r in result:
results.append(r)
return results
class ProcessSpiderOutputInvalidResult(BaseAsyncSpiderMiddlewareTestCase):
@defer.inlineCallbacks
def test_non_iterable(self):
with self.assertRaisesRegex(
_InvalidOutput,
(
r"\.process_spider_output must return an iterable, got <class "
r"'NoneType'>"
),
):
yield self._get_middleware_result(
ProcessSpiderOutputNonIterableMiddleware,
)
@defer.inlineCallbacks
def test_coroutine(self):
with self.assertRaisesRegex(
_InvalidOutput,
r"\.process_spider_output must be an asynchronous generator",
):
yield self._get_middleware_result(
ProcessSpiderOutputCoroutineMiddleware,
)
class ProcessStartRequestsSimpleMiddleware:
def process_start_requests(self, start_requests, spider):
for r in start_requests:
yield r
class ProcessStartRequestsSimple(BaseAsyncSpiderMiddlewareTestCase):
""" process_start_requests tests for simple start_requests"""
ITEM_TYPE = Request
MW_SIMPLE = ProcessStartRequestsSimpleMiddleware
def _start_requests(self):
for i in range(3):
yield Request(f'https://example.com/{i}', dont_filter=True)
@defer.inlineCallbacks
def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES_BASE': {}, 'SPIDER_MIDDLEWARES': setting})
self.spider = self.crawler._create_spider('foo')
self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
start_requests = iter(self._start_requests())
results = yield self.mwman.process_start_requests(start_requests, self.spider)
return results
def test_simple(self):
""" Simple mw """
return self._test_simple_base(self.MW_SIMPLE)
class UniversalMiddlewareNoSync:
async def process_spider_output_async(self, response, result, spider):
yield
class UniversalMiddlewareBothSync:
def process_spider_output(self, response, result, spider):
yield
def process_spider_output_async(self, response, result, spider):
yield
class UniversalMiddlewareBothAsync:
async def process_spider_output(self, response, result, spider):
yield
async def process_spider_output_async(self, response, result, spider):
yield
class UniversalMiddlewareManagerTest(TestCase):
def setUp(self):
self.mwman = SpiderMiddlewareManager()
def test_simple_mw(self):
mw = ProcessSpiderOutputSimpleMiddleware
self.mwman._add_middleware(mw)
self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
def test_async_mw(self):
mw = ProcessSpiderOutputAsyncGenMiddleware
self.mwman._add_middleware(mw)
self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
def test_universal_mw(self):
mw = ProcessSpiderOutputUniversalMiddleware
self.mwman._add_middleware(mw)
self.assertEqual(self.mwman.methods['process_spider_output'][0],
(mw.process_spider_output, mw.process_spider_output_async))
def test_universal_mw_no_sync(self):
with LogCapture() as log:
self.mwman._add_middleware(UniversalMiddlewareNoSync)
self.assertIn("UniversalMiddlewareNoSync has process_spider_output_async"
" without process_spider_output", str(log))
self.assertEqual(self.mwman.methods['process_spider_output'][0], None)
def test_universal_mw_both_sync(self):
mw = UniversalMiddlewareBothSync
with LogCapture() as log:
self.mwman._add_middleware(mw)
self.assertIn("UniversalMiddlewareBothSync.process_spider_output_async "
"is not an async generator function", str(log))
self.assertEqual(self.mwman.methods['process_spider_output'][0], mw.process_spider_output)
def test_universal_mw_both_async(self):
with LogCapture() as log:
self.mwman._add_middleware(UniversalMiddlewareBothAsync)
self.assertIn("UniversalMiddlewareBothAsync.process_spider_output "
"is an async generator function while process_spider_output_async exists",
str(log))
self.assertEqual(self.mwman.methods['process_spider_output'][0], None)
class BuiltinMiddlewareSimpleTest(BaseAsyncSpiderMiddlewareTestCase):
ITEM_TYPE = dict
MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
@defer.inlineCallbacks
def _get_middleware_result(self, *mw_classes, start_index: Optional[int] = None):
setting = self._construct_mw_setting(*mw_classes, start_index=start_index)
self.crawler = get_crawler(Spider, {'SPIDER_MIDDLEWARES': setting})
self.spider = self.crawler._create_spider('foo')
self.mwman = SpiderMiddlewareManager.from_crawler(self.crawler)
result = yield self.mwman.scrape_response(self._scrape_func, self.response, self.request, self.spider)
return result
def test_just_builtin(self):
return self._test_simple_base()
def test_builtin_simple(self):
return self._test_simple_base(self.MW_SIMPLE, start_index=1000)
def test_builtin_async(self):
""" Upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN, start_index=1000)
def test_builtin_universal(self):
return self._test_simple_base(self.MW_UNIVERSAL, start_index=1000)
def test_simple_builtin(self):
return self._test_simple_base(self.MW_SIMPLE)
def test_async_builtin(self):
""" Upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN)
def test_universal_builtin(self):
return self._test_simple_base(self.MW_UNIVERSAL)
class BuiltinMiddlewareAsyncGenTest(BuiltinMiddlewareSimpleTest):
async def _scrape_func(self, *args, **kwargs):
for item in super()._scrape_func():
yield item
def test_just_builtin(self):
return self._test_asyncgen_base()
def test_builtin_simple(self):
""" Downgrade """
return self._test_simple_base(self.MW_SIMPLE, downgrade=True, start_index=1000)
def test_builtin_async(self):
return self._test_asyncgen_base(self.MW_ASYNCGEN, start_index=1000)
def test_builtin_universal(self):
return self._test_asyncgen_base(self.MW_UNIVERSAL, start_index=1000)
def test_simple_builtin(self):
""" Downgrade """
return self._test_simple_base(self.MW_SIMPLE, downgrade=True)
def test_async_builtin(self):
return self._test_asyncgen_base(self.MW_ASYNCGEN)
def test_universal_builtin(self):
return self._test_asyncgen_base(self.MW_UNIVERSAL)
class ProcessSpiderExceptionTest(BaseAsyncSpiderMiddlewareTestCase):
ITEM_TYPE = dict
MW_SIMPLE = ProcessSpiderOutputSimpleMiddleware
MW_ASYNCGEN = ProcessSpiderOutputAsyncGenMiddleware
MW_UNIVERSAL = ProcessSpiderOutputUniversalMiddleware
MW_EXC_SIMPLE = ProcessSpiderExceptionSimpleIterableMiddleware
MW_EXC_ASYNCGEN = ProcessSpiderExceptionAsyncIterableMiddleware
def _scrape_func(self, *args, **kwargs):
1 / 0
@defer.inlineCallbacks
def _test_asyncgen_nodowngrade(self, *mw_classes):
with self.assertRaisesRegex(_InvalidOutput, "Async iterable returned from .+ cannot be downgraded"):
yield self._get_middleware_result(*mw_classes)
def test_exc_simple(self):
""" Simple exc mw """
return self._test_simple_base(self.MW_EXC_SIMPLE)
def test_exc_async(self):
""" Async exc mw """
return self._test_asyncgen_base(self.MW_EXC_ASYNCGEN)
def test_exc_simple_simple(self):
""" Simple exc mw -> simple output mw """
return self._test_simple_base(self.MW_SIMPLE,
self.MW_EXC_SIMPLE)
def test_exc_async_async(self):
""" Async exc mw -> async output mw """
return self._test_asyncgen_base(self.MW_ASYNCGEN,
self.MW_EXC_ASYNCGEN)
def test_exc_simple_async(self):
""" Simple exc mw -> async output mw; upgrade """
return self._test_asyncgen_base(self.MW_ASYNCGEN,
self.MW_EXC_SIMPLE)
def test_exc_async_simple(self):
""" Async exc mw -> simple output mw; cannot work as downgrading is not supported """
return self._test_asyncgen_nodowngrade(self.MW_SIMPLE,
self.MW_EXC_ASYNCGEN)

35
tests/test_spidermiddleware_output_chain.py
View File

@ -28,6 +28,7 @@ class RecoveryMiddleware:
class RecoverySpider(Spider):
name = 'RecoverySpider'
custom_settings = {
'SPIDER_MIDDLEWARES_BASE': {},
'SPIDER_MIDDLEWARES': {
RecoveryMiddleware: 10,
},
@ -43,6 +44,14 @@ class RecoverySpider(Spider):
raise TabError()
class RecoveryAsyncGenSpider(RecoverySpider):
name = 'RecoveryAsyncGenSpider'
async def parse(self, response):
for r in super().parse(response):
yield r
# ================================================================================
# (1) exceptions from a spider middleware's process_spider_input method
class FailProcessSpiderInputMiddleware:
@ -99,6 +108,13 @@ class GeneratorCallbackSpider(Spider):
raise ImportError()
class AsyncGeneratorCallbackSpider(GeneratorCallbackSpider):
async def parse(self, response):
yield {'test': 1}
yield {'test': 2}
raise ImportError()
# ================================================================================
# (2.1) exceptions from a spider callback (generator, middleware right after callback)
class GeneratorCallbackSpiderMiddlewareRightAfterSpider(GeneratorCallbackSpider):
@ -307,6 +323,16 @@ class TestSpiderMiddleware(TestCase):
self.assertEqual(str(log).count("Middleware: TabError exception caught"), 1)
self.assertIn("'item_scraped_count': 3", str(log))
@defer.inlineCallbacks
def test_recovery_asyncgen(self):
"""
Same as test_recovery but with an async callback.
"""
log = yield self.crawl_log(RecoveryAsyncGenSpider)
self.assertIn("Middleware: TabError exception caught", str(log))
self.assertEqual(str(log).count("Middleware: TabError exception caught"), 1)
self.assertIn("'item_scraped_count': 3", str(log))
@defer.inlineCallbacks
def test_process_spider_input_without_errback(self):
"""
@ -342,6 +368,15 @@ class TestSpiderMiddleware(TestCase):
self.assertIn("Middleware: ImportError exception caught", str(log2))
self.assertIn("'item_scraped_count': 2", str(log2))
@defer.inlineCallbacks
def test_async_generator_callback(self):
"""
Same as test_generator_callback but with an async callback.
"""
log2 = yield self.crawl_log(AsyncGeneratorCallbackSpider)
self.assertIn("Middleware: ImportError exception caught", str(log2))
self.assertIn("'item_scraped_count': 2", str(log2))
@defer.inlineCallbacks
def test_generator_callback_right_after_callback(self):
"""

20
tests/test_utils_asyncgen.py Normal file
View File

@ -0,0 +1,20 @@
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
class AsyncgenUtilsTest(unittest.TestCase):
@deferred_f_from_coro_f
async def test_as_async_generator(self):
ag = as_async_generator(range(42))
results = []
async for i in ag:
results.append(i)
self.assertEqual(results, list(range(42)))
@deferred_f_from_coro_f
async def test_collect_asyncgen(self):
ag = as_async_generator(range(42))
results = await collect_asyncgen(ag)
self.assertEqual(results, list(range(42)))

96
tests/test_utils_defer.py
View File

@ -1,12 +1,18 @@
import random
from pytest import mark
from twisted.trial import unittest
from twisted.internet import reactor, defer
from twisted.python.failure import Failure
from scrapy.utils.asyncgen import collect_asyncgen, as_async_generator
from scrapy.utils.defer import (
aiter_errback,
deferred_f_from_coro_f,
iter_errback,
maybe_deferred_to_future,
mustbe_deferred,
parallel_async,
process_chain,
process_chain_both,
process_parallel,
@ -121,6 +127,34 @@ class IterErrbackTest(unittest.TestCase):
self.assertIsInstance(errors[0].value, ZeroDivisionError)
class AiterErrbackTest(unittest.TestCase):
@deferred_f_from_coro_f
async def test_aiter_errback_good(self):
async def itergood():
for x in range(10):
yield x
errors = []
out = await collect_asyncgen(aiter_errback(itergood(), errors.append))
self.assertEqual(out, list(range(10)))
self.assertFalse(errors)
@deferred_f_from_coro_f
async def test_iter_errback_bad(self):
async def iterbad():
for x in range(10):
if x == 5:
1 / 0
yield x
errors = []
out = await collect_asyncgen(aiter_errback(iterbad(), errors.append))
self.assertEqual(out, [0, 1, 2, 3, 4])
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0].value, ZeroDivisionError)
class AsyncDefTestsuiteTest(unittest.TestCase):
@deferred_f_from_coro_f
async def test_deferred_f_from_coro_f(self):
@ -134,3 +168,65 @@ class AsyncDefTestsuiteTest(unittest.TestCase):
@deferred_f_from_coro_f
async def test_deferred_f_from_coro_f_xfail(self):
raise Exception("This is expected to be raised")
class AsyncCooperatorTest(unittest.TestCase):
""" This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
parallel_async is called with the results of a callback (so an iterable of items, requests and None,
with arbitrary delays between values), and it uses Scraper._process_spidermw_output as the callable
(so a callable that returns a Deferred for an item, which will fire after pipelines process it, and
None for everything else). The concurrent task count is the CONCURRENT_ITEMS setting.
We want to test different concurrency values compared to the iterable length.
We also want to simulate the real usage, with arbitrary delays between getting the values
from the iterable. We also want to simulate sync and async results from the callable.
"""
CONCURRENT_ITEMS = 50
@staticmethod
def callable(o, results):
if random.random() < 0.4:
# simulate async processing
dfd = defer.Deferred()
dfd.addCallback(lambda _: results.append(o))
delay = random.random() / 8
reactor.callLater(delay, dfd.callback, None)
return dfd
else:
# simulate trivial sync processing
results.append(o)
@staticmethod
def get_async_iterable(length):
# simulate a simple callback without delays between results
return as_async_generator(range(length))
@staticmethod
async def get_async_iterable_with_delays(length):
# simulate a callback with delays between some of the results
for i in range(length):
if random.random() < 0.1:
dfd = defer.Deferred()
delay = random.random() / 20
reactor.callLater(delay, dfd.callback, None)
await maybe_deferred_to_future(dfd)
yield i
@defer.inlineCallbacks
def test_simple(self):
for length in [20, 50, 100]:
results = []
ait = self.get_async_iterable(length)
dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
yield dl
self.assertEqual(list(range(length)), sorted(results))
@defer.inlineCallbacks
def test_delays(self):
for length in [20, 50, 100]:
results = []
ait = self.get_async_iterable_with_delays(length)
dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
yield dl
self.assertEqual(list(range(length)), sorted(results))

58
tests/test_utils_python.py
View File

@ -2,15 +2,18 @@ import functools
import gc
import operator
import platform
import unittest
from itertools import count
from warnings import catch_warnings, filterwarnings
from twisted.trial import unittest
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
from scrapy.utils.defer import deferred_f_from_coro_f, aiter_errback
from scrapy.utils.python import (
memoizemethod_noargs, binary_is_text, equal_attributes,
WeakKeyCache, get_func_args, to_bytes, to_unicode,
without_none_values, MutableChain)
without_none_values, MutableChain, MutableAsyncChain)
__doctests__ = ['scrapy.utils.python']
@ -32,6 +35,57 @@ class MutableChainTest(unittest.TestCase):
self.assertEqual(list(m), list(range(3, 13)))
class MutableAsyncChainTest(unittest.TestCase):
@staticmethod
async def g1():
for i in range(3):
yield i
@staticmethod
async def g2():
return
yield
@staticmethod
async def g3():
for i in range(7, 10):
yield i
@staticmethod
async def g4():
for i in range(3, 5):
yield i
1 / 0
for i in range(5, 7):
yield i
@staticmethod
async def collect_asyncgen_exc(asyncgen):
results = []
async for x in asyncgen:
results.append(x)
return results
@deferred_f_from_coro_f
async def test_mutableasyncchain(self):
m = MutableAsyncChain(self.g1(), as_async_generator(range(3, 7)))
m.extend(self.g2())
m.extend(self.g3())
self.assertEqual(await m.__anext__(), 0)
results = await collect_asyncgen(m)
self.assertEqual(results, list(range(1, 10)))
@deferred_f_from_coro_f
async def test_mutableasyncchain_exc(self):
m = MutableAsyncChain(self.g1())
m.extend(self.g4())
m.extend(self.g3())
results = await collect_asyncgen(aiter_errback(m, lambda _: None))
self.assertEqual(results, list(range(5)))
class ToUnicodeTest(unittest.TestCase):
def test_converting_an_utf8_encoded_string_to_unicode(self):
self.assertEqual(to_unicode(b'lel\xc3\xb1e'), 'lel\xf1e')

2
tests/test_utils_request.py
View File

@ -52,7 +52,7 @@ class UtilsRequestTest(unittest.TestCase):
class FingerprintTest(unittest.TestCase):
maxDiff = None
function = staticmethod(fingerprint)
function: staticmethod = staticmethod(fingerprint)
cache: Union[
"WeakKeyDictionary[Request, Dict[Tuple[Optional[Tuple[bytes, ...]], bool], bytes]]",
"WeakKeyDictionary[Request, Dict[Tuple[Optional[Tuple[bytes, ...]], bool], str]]",

2
tox.ini
View File

@ -38,7 +38,7 @@ install_command =
basepython = python3
deps =
lxml-stubs==0.2.0
mypy==0.910
mypy==0.971
types-pyOpenSSL==20.0.3
types-setuptools==57.0.0
commands =