Add parameters to some of typing.Callable.

This commit is contained in:
Andrey Rakhmatullin 2024-06-02 18:42:01 +05:00
parent 859a77ee42
commit 019f23e3b7
5 changed files with 25 additions and 15 deletions

View File

@ -1,5 +1,7 @@
"""Download handlers for different schemes"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, Union, cast
@ -19,16 +21,16 @@ logger = logging.getLogger(__name__)
class DownloadHandlers:
def __init__(self, crawler: "Crawler"):
self._crawler: "Crawler" = crawler
self._schemes: Dict[str, Union[str, Callable]] = (
def __init__(self, crawler: Crawler):
self._crawler: Crawler = crawler
self._schemes: Dict[str, Union[str, Callable[..., Any]]] = (
{}
) # stores acceptable schemes on instancing
self._handlers: Dict[str, Any] = {} # stores instanced handlers for schemes
self._notconfigured: Dict[str, str] = {} # remembers failed handlers
handlers: Dict[str, Union[str, Callable]] = without_none_values(
handlers: Dict[str, Union[str, Callable[..., Any]]] = without_none_values(
cast(
Dict[str, Union[str, Callable]],
Dict[str, Union[str, Callable[..., Any]]],
crawler.settings.getwithbase("DOWNLOAD_HANDLERS"),
)
)

View File

@ -86,7 +86,11 @@ class Slot:
class ExecutionEngine:
def __init__(self, crawler: Crawler, spider_closed_callback: Callable) -> None:
def __init__(
self,
crawler: Crawler,
spider_closed_callback: Callable[[Spider], Optional[Deferred[None]]],
) -> None:
self.crawler: Crawler = crawler
self.settings: Settings = crawler.settings
self.signals: SignalManager = crawler.signals
@ -102,7 +106,9 @@ class ExecutionEngine:
downloader_cls: Type[Downloader] = load_object(self.settings["DOWNLOADER"])
self.downloader: Downloader = downloader_cls(crawler)
self.scraper = Scraper(crawler)
self._spider_closed_callback: Callable = spider_closed_callback
self._spider_closed_callback: Callable[[Spider], Optional[Deferred[None]]] = (
spider_closed_callback
)
self.start_time: Optional[float] = None
def _get_scheduler_class(self, settings: BaseSettings) -> Type[BaseScheduler]:
@ -427,7 +433,7 @@ class ExecutionEngine:
dfd = self.slot.close()
def log_failure(msg: str) -> Callable:
def log_failure(msg: str) -> Callable[[Failure], None]:
def errback(failure: Failure) -> None:
logger.error(
msg, exc_info=failure_to_exc_info(failure), extra={"spider": spider}

View File

@ -266,7 +266,7 @@ class Request(object_ref):
return d
def _find_method(obj: Any, func: Callable) -> str:
def _find_method(obj: Any, func: Callable[..., Any]) -> str:
"""Helper function for Request.to_dict"""
# Only instance methods contain ``__func__``
if obj and hasattr(func, "__func__"):

View File

@ -56,7 +56,7 @@ def arg_to_iter(arg: Any) -> Iterable[Any]:
return [arg]
def load_object(path: Union[str, Callable]) -> Any:
def load_object(path: Union[str, Callable[..., Any]]) -> Any:
"""Load an object given its absolute object path, and return it.
The object can be the import path of a class, function, variable or an
@ -263,7 +263,7 @@ def walk_callable(node: ast.AST) -> Generator[ast.AST, Any, None]:
_generator_callbacks_cache = LocalWeakReferencedCache(limit=128)
def is_generator_with_return_value(callable: Callable) -> bool:
def is_generator_with_return_value(callable: Callable[..., Any]) -> bool:
"""
Returns True if a callable is a generator function which includes a
'return' statement with a value different than None, False otherwise
@ -300,7 +300,9 @@ def is_generator_with_return_value(callable: Callable) -> bool:
return bool(_generator_callbacks_cache[callable])
def warn_on_generator_with_return_value(spider: Spider, callable: Callable) -> None:
def warn_on_generator_with_return_value(
spider: Spider, callable: Callable[..., Any]
) -> None:
"""
Logs a warning if a callable is a generator function and includes
a 'return' statement with a value different than None

View File

@ -217,7 +217,7 @@ def binary_is_text(data: bytes) -> bool:
return all(c not in _BINARYCHARS for c in data)
def get_func_args(func: Callable, stripself: bool = False) -> List[str]:
def get_func_args(func: Callable[..., Any], stripself: bool = False) -> List[str]:
"""Return the argument name list of a callable object"""
if not callable(func):
raise TypeError(f"func must be callable, got '{type(func).__name__}'")
@ -247,7 +247,7 @@ def get_func_args(func: Callable, stripself: bool = False) -> List[str]:
return args
def get_spec(func: Callable) -> Tuple[List[str], Dict[str, Any]]:
def get_spec(func: Callable[..., Any]) -> Tuple[List[str], Dict[str, Any]]:
"""Returns (args, kwargs) tuple for a function
>>> import re
>>> get_spec(re.match)
@ -285,7 +285,7 @@ def get_spec(func: Callable) -> Tuple[List[str], Dict[str, Any]]:
def equal_attributes(
obj1: Any, obj2: Any, attributes: Optional[List[Union[str, Callable]]]
obj1: Any, obj2: Any, attributes: Optional[List[Union[str, Callable[[Any], Any]]]]
) -> bool:
"""Compare two objects attributes"""
# not attributes given return False by default