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

Add/update typing, cleanup iterator/iterable inconsistencies.

This commit is contained in:
Andrey Rakhmatullin 2021-04-14 18:21:58 +05:00
parent ffc6f525ce
commit 61197d3dba
5 changed files with 71 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
scrapy/core/scraper.py
View File

@ -1,10 +1,8 @@
"""This module implements the Scraper component which parses responses and
extracts information from them"""
import collections
import logging
from collections import deque
from collections.abc import Iterable
from typing import Union
from typing import AsyncGenerator, AsyncIterable, Generator, Iterable, Union
from itemadapter import is_item
from twisted.internet import defer
@ -188,7 +186,8 @@ class Scraper:
def handle_spider_output(self, result: Iterable, request: Request, response: Response, spider: Spider):
if not result:
return defer_succeed(None)
if isinstance(result, collections.abc.AsyncIterable):
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)

36
scrapy/core/spidermw.py
View File

@ -3,9 +3,8 @@ Spider Middleware manager
See documentation in docs/topics/spider-middleware.rst
"""
import collections.abc
from itertools import islice
from typing import Any, Callable, Generator, Iterable, Union, AsyncIterable
from typing import Any, Callable, Generator, Iterable, Union, AsyncIterable, AsyncGenerator
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
@ -61,8 +60,9 @@ 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:
def _evaluate_iterable(self, response: Response, spider: Spider, iterable: Union[Iterable, AsyncIterable],
exception_processor_index: int, recover_to: Union[MutableChain, MutableAsyncChain]
) -> Union[Generator, AsyncGenerator]:
@_process_iterable_universal
async def _evaluate_async_iterable(iterable):
try:
@ -100,11 +100,13 @@ class SpiderMiddlewareManager(MiddlewareManager):
return _failure
def _process_spider_output(self, response: Response, spider: Spider,
result: Iterable, start_index: int = 0) -> MutableChain:
result: Union[Iterable, AsyncIterable], start_index: int = 0
) -> Union[MutableChain, MutableAsyncChain]:
# 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
last_result_async = isinstance(result, collections.abc.AsyncIterator)
if last_result_async:
recovered: Union[MutableChain, MutableAsyncChain]
last_result_is_async = isinstance(result, AsyncIterable)
if last_result_is_async:
recovered = MutableAsyncChain()
else:
recovered = MutableChain()
@ -127,30 +129,32 @@ class SpiderMiddlewareManager(MiddlewareManager):
msg = (f"Middleware {method.__qualname__} must return an "
f"iterable, got {type(result)}")
raise _InvalidOutput(msg)
if last_result_async and isinstance(result, collections.abc.Iterator):
if last_result_is_async and isinstance(result, Iterable):
raise TypeError(f"Synchronous {method.__qualname__} called with an async iterable")
last_result_async = isinstance(result, collections.abc.AsyncIterator)
last_result_is_async = isinstance(result, AsyncIterable)
if last_result_async:
if last_result_is_async:
return MutableAsyncChain(result, recovered)
else:
return MutableChain(result, recovered)
return MutableChain(result, recovered) # type: ignore[arg-type]
def _process_callback_output(self, response: Response, spider: Spider, result: Iterable) -> MutableChain:
if isinstance(result, collections.abc.AsyncIterator):
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)
result = self._process_spider_output(response, spider, result)
if isinstance(result, collections.abc.AsyncIterator):
if isinstance(result, AsyncIterable):
return MutableAsyncChain(result, recovered)
else:
return MutableChain(result, recovered)
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:
def process_callback_output(result: Union[Iterable, AsyncIterable]) -> Union[MutableChain, MutableAsyncChain]:
return self._process_callback_output(response, spider, result)
def process_spider_exception(_failure: Failure) -> Union[Failure, MutableChain]:

19
scrapy/utils/asyncgen.py
View File

@ -1,7 +1,6 @@
import collections
import functools
import inspect
from collections.abc import AsyncIterable
from typing import AsyncGenerator, AsyncIterable, Callable, Generator, Iterable, Union
async def collect_asyncgen(result: AsyncIterable):
@ -11,9 +10,9 @@ async def collect_asyncgen(result: AsyncIterable):
return results
async def as_async_generator(it):
""" Wraps an iterator (sync or async) into an async generator. """
if isinstance(it, collections.abc.AsyncIterator):
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:
@ -22,7 +21,7 @@ async def as_async_generator(it):
# https://stackoverflow.com/a/66170760/113586
def _process_iterable_universal(process_async):
def _process_iterable_universal(process_async: Callable):
""" Takes a function that takes an async iterable, args and kwargs. Returns
a function that takes any iterable, args and kwargs.
@ -33,7 +32,7 @@ def _process_iterable_universal(process_async):
# If this stops working, all internal uses can be just replaced with manually-written
# process_sync functions.
def process_sync(iterable, *args, **kwargs):
def process_sync(iterable: Iterable, *args, **kwargs) -> Generator:
agen = process_async(as_async_generator(iterable), *args, **kwargs)
if not inspect.isasyncgen(agen):
raise ValueError(f"process_async returned wrong type {type(agen)}")
@ -52,11 +51,11 @@ def _process_iterable_universal(process_async):
f"you can't use {_process_iterable_universal.__name__} with it.")
@functools.wraps(process_async)
def process(iterable, *args, **kwargs):
if inspect.isasyncgen(iterable):
def process(iterable: Union[Iterable, AsyncIterable], *args, **kwargs) -> Union[Generator, AsyncGenerator]:
if isinstance(iterable, AsyncIterable):
# call process_async directly
return process_async(iterable, *args, **kwargs)
if hasattr(iterable, '__iter__'):
if isinstance(iterable, Iterable):
# convert process_async to process_sync
return process_sync(iterable, *args, **kwargs)
raise TypeError(f"Wrong iterable type {type(iterable)}")

52
scrapy/utils/defer.py
View File

@ -3,9 +3,21 @@ Helper functions for dealing with Twisted deferreds
"""
import asyncio
import inspect
from collections.abc import Coroutine
from asyncio import Future
from functools import wraps
from typing import Any, Callable, Generator, Iterable
from typing import (
Any,
AsyncGenerator,
AsyncIterable,
Callable,
Coroutine,
Generator,
Iterable,
Iterator,
List,
Optional,
Union
)
from twisted.internet import defer
from twisted.internet.defer import Deferred, DeferredList, ensureDeferred
@ -80,8 +92,8 @@ def parallel(iterable: Iterable, count: int, callable: Callable, *args, **named)
return DeferredList([coop.coiterate(work) for _ in range(count)])
class _AsyncCooperatorAdapter:
""" A class that wraps an async iterator into a normal iterator suitable
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.
@ -125,30 +137,30 @@ class _AsyncCooperatorAdapter:
Cooperator/CooperativeTask and use it instead of this adapter to achieve the same
goal.
"""
def __init__(self, aiterator, callable, *callable_args, **callable_kwargs):
self.aiterator = aiterator
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 = []
self.anext_deferred = None
self.waiting_deferreds: List[Deferred] = []
self.anext_deferred: Optional[Deferred] = None
def _callback(self, result):
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, defer.Deferred):
if isinstance(result, Deferred):
result.chainDeferred(d)
else:
d.callback(None)
if self.waiting_deferreds:
self._call_anext()
def _errback(self, failure):
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
@ -157,29 +169,29 @@ class _AsyncCooperatorAdapter:
for d in self.waiting_deferreds:
d.callback(None)
def _call_anext(self):
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):
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 = defer.Deferred()
d = Deferred()
self.waiting_deferreds.append(d)
if not self.anext_deferred:
self._call_anext()
return d
def parallel_async(async_iterable, count, callable, *args, **named):
""" Like parallel but for async iterables """
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 = defer.DeferredList([coop.coiterate(work) for _ in range(count)])
dl = DeferredList([coop.coiterate(work) for _ in range(count)])
return dl
@ -232,7 +244,7 @@ def iter_errback(iterable: Iterable, errback: Callable, *a, **kw) -> Generator:
errback(failure.Failure(), *a, **kw)
async def aiter_errback(aiterable, errback, *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()
"""
@ -290,13 +302,13 @@ def maybeDeferred_coro(f: Callable, *args, **kw) -> Deferred:
return defer.succeed(result)
def deferred_to_future(d):
def deferred_to_future(d: Deferred) -> Future:
""" Wraps a Deferred into a Future. Requires the asyncio reactor.
"""
return d.asFuture(asyncio.get_event_loop())
def maybe_deferred_to_future(d):
def maybe_deferred_to_future(d: Deferred) -> Union[Deferred, Future]:
""" Converts a Deferred to something that can be awaited in a callback or other user coroutine.
If the asyncio reactor is installed, coroutines are wrapped into Futures, and only Futures can be

14
scrapy/utils/python.py
View File

@ -8,9 +8,9 @@ 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 AsyncIterable, Iterable, Union
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.asyncgen import as_async_generator
@ -345,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):
@ -359,22 +359,22 @@ class MutableChain(Iterable):
return self.__next__()
async def _async_chain(*iterables):
async def _async_chain(*iterables: Union[Iterable, AsyncIterable]):
for it in iterables:
async for o in as_async_generator(it):
yield o
class MutableAsyncChain:
class MutableAsyncChain(AsyncIterable):
"""
Similar to MutableChain but for async iterables
"""
def __init__(self, *args):
def __init__(self, *args: Union[Iterable, AsyncIterable]):
self.data = _async_chain(*args)
def extend(self, *aiterables):
self.data = _async_chain(self.data, _async_chain(*aiterables))
def extend(self, *iterables: Union[Iterable, AsyncIterable]) -> None:
self.data = _async_chain(self.data, _async_chain(*iterables))
def __aiter__(self):
return self