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

Use ParamSpec for callables. (#6353)

This commit is contained in:
Andrey Rakhmatullin 2024-05-13 14:01:52 +04:00 committed by GitHub
parent 93f0628530
commit 4ed5c5ae91
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 204 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
scrapy/cmdline.py
View File

@ -1,10 +1,12 @@
from __future__ import annotations
import argparse
import cProfile
import inspect
import os
import sys
from importlib.metadata import entry_points
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Type
from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Optional, Tuple, Type
import scrapy
from scrapy.commands import BaseRunSpiderCommand, ScrapyCommand, ScrapyHelpFormatter
@ -15,6 +17,12 @@ from scrapy.utils.misc import walk_modules
from scrapy.utils.project import get_project_settings, inside_project
from scrapy.utils.python import garbage_collect
if TYPE_CHECKING:
# typing.ParamSpec requires Python 3.10
from typing_extensions import ParamSpec
_P = ParamSpec("_P")
class ScrapyArgumentParser(argparse.ArgumentParser):
def _parse_optional(
@ -121,7 +129,10 @@ def _print_unknown_command(
def _run_print_help(
parser: argparse.ArgumentParser, func: Callable, *a: Any, **kw: Any
parser: argparse.ArgumentParser,
func: Callable[_P, None],
*a: _P.args,
**kw: _P.kwargs,
) -> None:
try:
func(*a, **kw)

10
scrapy/core/downloader/handlers/ftp.py
View File

@ -109,12 +109,10 @@ class FTPDownloadHandler:
def gotClient(self, client: FTPClient, request: Request, filepath: str) -> Deferred:
self.client = client
protocol = ReceivedDataProtocol(request.meta.get("ftp_local_filename"))
return client.retrieveFile(filepath, protocol).addCallbacks(
callback=self._build_response,
callbackArgs=(request, protocol),
errback=self._failed,
errbackArgs=(request,),
)
d = client.retrieveFile(filepath, protocol)
d.addCallback(self._build_response, request, protocol)
d.addErrback(self._failed, request)
return d
def _build_response(
self, result: Any, request: Request, protocol: ReceivedDataProtocol

2
scrapy/core/engine.py
View File

@ -347,7 +347,7 @@ class ExecutionEngine:
assert self.spider is not None
dwld = self.downloader.fetch(request, self.spider)
dwld.addCallbacks(_on_success)
dwld.addCallback(_on_success)
dwld.addBoth(_on_complete)
return dwld

10
scrapy/core/scraper.py
View File

@ -8,7 +8,6 @@ from collections import deque
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
AsyncIterable,
Deque,
Generator,
@ -18,6 +17,7 @@ from typing import (
Tuple,
Type,
Union,
cast,
)
from itemadapter import is_item
@ -184,7 +184,9 @@ class Scraper:
result, request, spider
) # returns spider's processed output
dfd.addErrback(self.handle_spider_error, request, result, spider)
dfd.addCallback(self.handle_spider_output, request, result, spider)
dfd.addCallback(
self.handle_spider_output, request, cast(Response, result), spider
)
return dfd
def _scrape2(
@ -256,12 +258,12 @@ class Scraper:
self,
result: Union[Iterable, AsyncIterable],
request: Request,
response: Union[Response, Failure],
response: Response,
spider: Spider,
) -> Deferred:
if not result:
return defer_succeed(None)
it: Union[Generator, AsyncGenerator]
it: Union[Iterable, AsyncIterable]
if isinstance(result, AsyncIterable):
it = aiter_errback(
result, self.handle_spider_error, request, response, spider

6
scrapy/core/spidermw.py
View File

@ -303,10 +303,8 @@ class SpiderMiddlewareManager(MiddlewareManager):
dfd = mustbe_deferred(
self._process_spider_input, scrape_func, response, request, spider
)
dfd.addCallbacks(
callback=deferred_f_from_coro_f(process_callback_output),
errback=process_spider_exception,
)
dfd.addCallback(deferred_f_from_coro_f(process_callback_output))
dfd.addErrback(process_spider_exception)
return dfd
def process_start_requests(

8
scrapy/mail.py
View File

@ -154,12 +154,8 @@ class MailSender:
return None
dfd = self._sendmail(rcpts, msg.as_string().encode(charset or "utf-8"))
dfd.addCallbacks(
callback=self._sent_ok,
errback=self._sent_failed,
callbackArgs=(to, cc, subject, len(attachs)),
errbackArgs=(to, cc, subject, len(attachs)),
)
dfd.addCallback(self._sent_ok, to, cc, subject, len(attachs))
dfd.addErrback(self._sent_failed, to, cc, subject, len(attachs))
reactor.addSystemEventTrigger("before", "shutdown", lambda: dfd)
return dfd

3
scrapy/pipelines/files.py
View File

@ -459,7 +459,8 @@ class FilesPipeline(MediaPipeline):
path = self.file_path(request, info=info, item=item)
dfd = defer.maybeDeferred(self.store.stat_file, path, info)
dfd.addCallbacks(_onsuccess, lambda _: None)
dfd.addCallback(_onsuccess)
dfd.addErrback(lambda _: None)
dfd.addErrback(
lambda f: logger.error(
self.__class__.__name__ + ".store.stat_file",

27
scrapy/pipelines/media.py
View File

@ -106,10 +106,17 @@ class MediaPipeline:
# Return cached result if request was already seen
if fp in info.downloaded:
return defer_result(info.downloaded[fp]).addCallbacks(cb, eb)
d = defer_result(info.downloaded[fp])
d.addCallback(cb)
if eb:
d.addErrback(eb)
return d
# Otherwise, wait for result
wad = Deferred().addCallbacks(cb, eb)
wad = Deferred()
wad.addCallback(cb)
if eb:
wad.addErrback(eb)
info.waiting[fp].append(wad)
# Check if request is downloading right now to avoid doing it twice
@ -140,23 +147,11 @@ class MediaPipeline:
if self.download_func:
# this ugly code was left only to support tests. TODO: remove
dfd = mustbe_deferred(self.download_func, request, info.spider)
dfd.addCallbacks(
callback=self.media_downloaded,
callbackArgs=(request, info),
callbackKeywords={"item": item},
errback=self.media_failed,
errbackArgs=(request, info),
)
else:
self._modify_media_request(request)
dfd = self.crawler.engine.download(request)
dfd.addCallbacks(
callback=self.media_downloaded,
callbackArgs=(request, info),
callbackKeywords={"item": item},
errback=self.media_failed,
errbackArgs=(request, info),
)
dfd.addCallback(self.media_downloaded, request, info, item=item)
dfd.addErrback(self.media_failed, request, info)
return dfd
def _cache_result_and_execute_waiters(self, result, fp, info):

4
scrapy/shell.py
View File

@ -231,7 +231,9 @@ def _request_deferred(request: Request) -> defer.Deferred:
d: defer.Deferred = defer.Deferred()
d.addBoth(_restore_callbacks)
if request.callback:
d.addCallbacks(request.callback, request.errback)
d.addCallback(request.callback)
if request.errback:
d.addErrback(request.errback)
request.callback, request.errback = d.callback, d.errback
return d

29
scrapy/utils/decorators.py
View File

@ -1,21 +1,34 @@
from __future__ import annotations
import warnings
from functools import wraps
from typing import Any, Callable
from typing import TYPE_CHECKING, Any, Callable, TypeVar
from twisted.internet import defer, threads
from twisted.internet.defer import Deferred
from scrapy.exceptions import ScrapyDeprecationWarning
if TYPE_CHECKING:
# typing.ParamSpec requires Python 3.10
from typing_extensions import ParamSpec
def deprecated(use_instead: Any = None) -> Callable:
_P = ParamSpec("_P")
_T = TypeVar("_T")
def deprecated(
use_instead: Any = None,
) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used."""
def deco(func: Callable) -> Callable:
def deco(func: Callable[_P, _T]) -> Callable[_P, _T]:
@wraps(func)
def wrapped(*args: Any, **kwargs: Any) -> Any:
def wrapped(*args: _P.args, **kwargs: _P.kwargs) -> Any:
message = f"Call to deprecated function {func.__name__}."
if use_instead:
message += f" Use {use_instead} instead."
@ -30,23 +43,23 @@ def deprecated(use_instead: Any = None) -> Callable:
return deco
def defers(func: Callable) -> Callable[..., Deferred]:
def defers(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:
"""Decorator to make sure a function always returns a deferred"""
@wraps(func)
def wrapped(*a: Any, **kw: Any) -> Deferred:
def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
return defer.maybeDeferred(func, *a, **kw)
return wrapped
def inthread(func: Callable) -> Callable[..., Deferred]:
def inthread(func: Callable[_P, _T]) -> Callable[_P, Deferred[_T]]:
"""Decorator to call a function in a thread and return a deferred with the
result
"""
@wraps(func)
def wrapped(*a: Any, **kw: Any) -> Deferred:
def wrapped(*a: _P.args, **kw: _P.kwargs) -> Deferred[_T]:
return threads.deferToThread(func, *a, **kw)
return wrapped

135
scrapy/utils/defer.py
View File

@ -2,21 +2,22 @@
Helper functions for dealing with Twisted deferreds
"""
from __future__ import annotations
import asyncio
import inspect
from asyncio import Future
from functools import wraps
from types import CoroutineType
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
AsyncIterable,
AsyncIterator,
Awaitable,
Callable,
Coroutine,
Dict,
Generator,
Iterable,
Iterator,
List,
@ -37,6 +38,14 @@ from twisted.python.failure import Failure
from scrapy.exceptions import IgnoreRequest
from scrapy.utils.reactor import _get_asyncio_event_loop, is_asyncio_reactor_installed
if TYPE_CHECKING:
# typing.Concatenate and typing.ParamSpec require Python 3.10
from typing_extensions import Concatenate, ParamSpec
_P = ParamSpec("_P")
_T = TypeVar("_T")
def defer_fail(_failure: Failure) -> Deferred:
"""Same as twisted.internet.defer.fail but delay calling errback until
@ -74,7 +83,31 @@ def defer_result(result: Any) -> Deferred:
return defer_succeed(result)
def mustbe_deferred(f: Callable, *args: Any, **kw: Any) -> Deferred:
@overload
def mustbe_deferred(
f: Callable[_P, Deferred[_T]], *args: _P.args, **kw: _P.kwargs
) -> Deferred[_T]: ...
@overload
def mustbe_deferred(
f: Callable[_P, Coroutine[Deferred[Any], Any, _T]],
*args: _P.args,
**kw: _P.kwargs,
) -> Deferred[_T]: ...
@overload
def mustbe_deferred(
f: Callable[_P, _T], *args: _P.args, **kw: _P.kwargs
) -> Deferred[_T]: ...
def mustbe_deferred(
f: Callable[_P, Union[Deferred[_T], Coroutine[Deferred[Any], Any, _T], _T]],
*args: _P.args,
**kw: _P.kwargs,
) -> Deferred[_T]:
"""Same as twisted.internet.defer.maybeDeferred, but delay calling
callback/errback to next reactor loop
"""
@ -92,7 +125,11 @@ def mustbe_deferred(f: Callable, *args: Any, **kw: Any) -> Deferred:
def parallel(
iterable: Iterable, count: int, callable: Callable, *args: Any, **named: Any
iterable: Iterable[_T],
count: int,
callable: Callable[Concatenate[_T, _P], Any],
*args: _P.args,
**named: _P.kwargs,
) -> Deferred:
"""Execute a callable over the objects in the given iterable, in parallel,
using no more than ``count`` concurrent calls.
@ -104,7 +141,7 @@ def parallel(
return DeferredList([coop.coiterate(work) for _ in range(count)])
class _AsyncCooperatorAdapter(Iterator):
class _AsyncCooperatorAdapter(Iterator[Deferred]):
"""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
@ -152,28 +189,30 @@ class _AsyncCooperatorAdapter(Iterator):
def __init__(
self,
aiterable: AsyncIterable,
callable: Callable,
*callable_args: Any,
**callable_kwargs: Any,
aiterable: AsyncIterable[_T],
callable: Callable[Concatenate[_T, _P], Any],
*callable_args: _P.args,
**callable_kwargs: _P.kwargs,
):
self.aiterator: AsyncIterator = aiterable.__aiter__()
self.callable: Callable = callable
self.aiterator: AsyncIterator[_T] = aiterable.__aiter__()
self.callable: Callable[Concatenate[_T, _P], Any] = callable
self.callable_args: Tuple[Any, ...] = callable_args
self.callable_kwargs: Dict[str, Any] = callable_kwargs
self.finished: bool = False
self.waiting_deferreds: List[Deferred] = []
self.anext_deferred: Optional[Deferred] = None
self.anext_deferred: Optional[Deferred[_T]] = None
def _callback(self, result: Any) -> None:
def _callback(self, result: _T) -> 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)
callable_result = self.callable(
result, *self.callable_args, **self.callable_kwargs
)
d = self.waiting_deferreds.pop(0)
if isinstance(result, Deferred):
result.chainDeferred(d)
if isinstance(callable_result, Deferred):
callable_result.chainDeferred(d)
else:
d.callback(None)
if self.waiting_deferreds:
@ -207,11 +246,11 @@ class _AsyncCooperatorAdapter(Iterator):
def parallel_async(
async_iterable: AsyncIterable,
async_iterable: AsyncIterable[_T],
count: int,
callable: Callable,
*args: Any,
**named: Any,
callable: Callable[Concatenate[_T, _P], Any],
*args: _P.args,
**named: _P.kwargs,
) -> Deferred:
"""Like parallel but for async iterators"""
coop = Cooperator()
@ -221,7 +260,10 @@ def parallel_async(
def process_chain(
callbacks: Iterable[Callable], input: Any, *a: Any, **kw: Any
callbacks: Iterable[Callable[Concatenate[_T, _P], Any]],
input: Any,
*a: _P.args,
**kw: _P.kwargs,
) -> Deferred:
"""Return a Deferred built by chaining the given callbacks"""
d: Deferred = Deferred()
@ -232,23 +274,17 @@ def process_chain(
def process_chain_both(
callbacks: Iterable[Callable],
errbacks: Iterable[Callable],
callbacks: Iterable[Callable[Concatenate[_T, _P], Any]],
errbacks: Iterable[Callable[Concatenate[Failure, _P], Any]],
input: Any,
*a: Any,
**kw: Any,
*a: _P.args,
**kw: _P.kwargs,
) -> Deferred:
"""Return a Deferred built by chaining the given callbacks and errbacks"""
d: Deferred = Deferred()
for cb, eb in zip(callbacks, errbacks):
d.addCallbacks(
callback=cb,
errback=eb,
callbackArgs=a,
callbackKeywords=kw,
errbackArgs=a,
errbackKeywords=kw,
)
d.addCallback(cb, *a, **kw)
d.addErrback(eb, *a, **kw)
if isinstance(input, failure.Failure):
d.errback(input)
else:
@ -257,20 +293,27 @@ def process_chain_both(
def process_parallel(
callbacks: Iterable[Callable], input: Any, *a: Any, **kw: Any
callbacks: Iterable[Callable[Concatenate[_T, _P], Any]],
input: Any,
*a: _P.args,
**kw: _P.kwargs,
) -> Deferred:
"""Return a Deferred with the output of all successful calls to the given
callbacks
"""
dfds = [defer.succeed(input).addCallback(x, *a, **kw) for x in callbacks]
d: Deferred = DeferredList(dfds, fireOnOneErrback=True, consumeErrors=True)
d.addCallbacks(lambda r: [x[1] for x in r], lambda f: f.value.subFailure)
d.addCallback(lambda r: [x[1] for x in r])
d.addErrback(lambda f: f.value.subFailure)
return d
def iter_errback(
iterable: Iterable, errback: Callable, *a: Any, **kw: Any
) -> Generator:
iterable: Iterable[_T],
errback: Callable[Concatenate[Failure, _P], Any],
*a: _P.args,
**kw: _P.kwargs,
) -> Iterable[_T]:
"""Wraps an iterable calling an errback if an error is caught while
iterating it.
"""
@ -285,8 +328,11 @@ def iter_errback(
async def aiter_errback(
aiterable: AsyncIterable, errback: Callable, *a: Any, **kw: Any
) -> AsyncGenerator:
aiterable: AsyncIterable[_T],
errback: Callable[Concatenate[Failure, _P], Any],
*a: _P.args,
**kw: _P.kwargs,
) -> AsyncIterable[_T]:
"""Wraps an async iterable calling an errback if an error is caught while
iterating it. Similar to scrapy.utils.defer.iter_errback()
"""
@ -301,7 +347,6 @@ async def aiter_errback(
_CT = TypeVar("_CT", bound=Union[Awaitable, CoroutineType, Future])
_T = TypeVar("_T")
@overload
@ -327,7 +372,9 @@ def deferred_from_coro(o: _T) -> Union[Deferred, _T]:
return o
def deferred_f_from_coro_f(coro_f: Callable[..., Coroutine]) -> Callable:
def deferred_f_from_coro_f(
coro_f: Callable[_P, Coroutine[Any, Any, _T]]
) -> Callable[_P, Deferred[_T]]:
"""Converts a coroutine function into a function that returns a Deferred.
The coroutine function will be called at the time when the wrapper is called. Wrapper args will be passed to it.
@ -335,13 +382,15 @@ def deferred_f_from_coro_f(coro_f: Callable[..., Coroutine]) -> Callable:
"""
@wraps(coro_f)
def f(*coro_args: Any, **coro_kwargs: Any) -> Any:
def f(*coro_args: _P.args, **coro_kwargs: _P.kwargs) -> Any:
return deferred_from_coro(coro_f(*coro_args, **coro_kwargs))
return f
def maybeDeferred_coro(f: Callable, *args: Any, **kw: Any) -> Deferred:
def maybeDeferred_coro(
f: Callable[_P, Any], *args: _P.args, **kw: _P.kwargs
) -> Deferred:
"""Copy of defer.maybeDeferred that also converts coroutines to Deferreds."""
try:
result = f(*args, **kw)

23
scrapy/utils/python.py
View File

@ -2,6 +2,8 @@
This module contains essential stuff that should've come with Python itself ;)
"""
from __future__ import annotations
import collections.abc
import gc
import inspect
@ -11,6 +13,7 @@ import weakref
from functools import partial, wraps
from itertools import chain
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
AsyncIterable,
@ -25,12 +28,21 @@ from typing import (
Optional,
Pattern,
Tuple,
TypeVar,
Union,
overload,
)
from scrapy.utils.asyncgen import as_async_generator
if TYPE_CHECKING:
# typing.Concatenate and typing.ParamSpec require Python 3.10
from typing_extensions import Concatenate, ParamSpec
_P = ParamSpec("_P")
_T = TypeVar("_T")
def flatten(x: Iterable) -> list:
"""flatten(sequence) -> list
@ -169,14 +181,19 @@ def re_rsearch(
return None
def memoizemethod_noargs(method: Callable) -> Callable:
_SelfT = TypeVar("_SelfT")
def memoizemethod_noargs(
method: Callable[Concatenate[_SelfT, _P], _T]
) -> Callable[Concatenate[_SelfT, _P], _T]:
"""Decorator to cache the result of a method (without arguments) using a
weak reference to its object
"""
cache: weakref.WeakKeyDictionary[Any, Any] = weakref.WeakKeyDictionary()
cache: weakref.WeakKeyDictionary[_SelfT, _T] = weakref.WeakKeyDictionary()
@wraps(method)
def new_method(self: Any, *args: Any, **kwargs: Any) -> Any:
def new_method(self: _SelfT, *args: _P.args, **kwargs: _P.kwargs) -> _T:
if self not in cache:
cache[self] = method(self, *args, **kwargs)
return cache[self]

33
scrapy/utils/reactor.py
View File

@ -1,8 +1,21 @@
from __future__ import annotations
import asyncio
import sys
from asyncio import AbstractEventLoop, AbstractEventLoopPolicy
from contextlib import suppress
from typing import Any, Callable, Dict, List, Optional, Sequence, Type
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Generic,
List,
Optional,
Tuple,
Type,
TypeVar,
)
from warnings import catch_warnings, filterwarnings, warn
from twisted.internet import asyncioreactor, error
@ -13,6 +26,14 @@ from twisted.internet.tcp import Port
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.misc import load_object
if TYPE_CHECKING:
# typing.ParamSpec requires Python 3.10
from typing_extensions import ParamSpec
_P = ParamSpec("_P")
_T = TypeVar("_T")
def listen_tcp(portrange: List[int], host: str, factory: ServerFactory) -> Port: # type: ignore[return]
"""Like reactor.listenTCP but tries different ports in a range."""
@ -32,14 +53,14 @@ def listen_tcp(portrange: List[int], host: str, factory: ServerFactory) -> Port:
raise
class CallLaterOnce:
class CallLaterOnce(Generic[_T]):
"""Schedule a function to be called in the next reactor loop, but only if
it hasn't been already scheduled since the last time it ran.
"""
def __init__(self, func: Callable, *a: Any, **kw: Any):
self._func: Callable = func
self._a: Sequence[Any] = a
def __init__(self, func: Callable[_P, _T], *a: _P.args, **kw: _P.kwargs):
self._func: Callable[_P, _T] = func
self._a: Tuple[Any, ...] = a
self._kw: Dict[str, Any] = kw
self._call: Optional[DelayedCall] = None
@ -53,7 +74,7 @@ class CallLaterOnce:
if self._call:
self._call.cancel()
def __call__(self) -> Any:
def __call__(self) -> _T:
self._call = None
return self._func(*self._a, **self._kw)