diff --git a/scrapy/commands/__init__.py b/scrapy/commands/__init__.py index 9baee3a48..2aa569cdd 100644 --- a/scrapy/commands/__init__.py +++ b/scrapy/commands/__init__.py @@ -4,7 +4,7 @@ Base class for Scrapy commands import argparse import os from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from twisted.python import failure @@ -116,7 +116,7 @@ class ScrapyCommand: if opts.pdb: failure.startDebugMode() - def run(self, args, opts): + def run(self, args: List[str], opts: argparse.Namespace) -> None: """ Entry point for running commands """ diff --git a/scrapy/commands/fetch.py b/scrapy/commands/fetch.py index 1359e445f..cdb7ad4ae 100644 --- a/scrapy/commands/fetch.py +++ b/scrapy/commands/fetch.py @@ -1,7 +1,10 @@ import sys +from argparse import Namespace +from typing import List, Type from w3lib.url import is_url +from scrapy import Spider from scrapy.commands import ScrapyCommand from scrapy.exceptions import UsageError from scrapy.http import Request @@ -57,7 +60,7 @@ class Command(ScrapyCommand): def _print_bytes(self, bytes_): sys.stdout.buffer.write(bytes_ + b"\n") - def run(self, args, opts): + def run(self, args: List[str], opts: Namespace) -> None: if len(args) != 1 or not is_url(args[0]): raise UsageError() request = Request( @@ -73,7 +76,8 @@ class Command(ScrapyCommand): else: request.meta["handle_httpstatus_all"] = True - spidercls = DefaultSpider + spidercls: Type[Spider] = DefaultSpider + assert self.crawler_process spider_loader = self.crawler_process.spider_loader if opts.spider: spidercls = spider_loader.load(opts.spider) diff --git a/scrapy/commands/shell.py b/scrapy/commands/shell.py index 63c23d04c..0a5e61f7a 100644 --- a/scrapy/commands/shell.py +++ b/scrapy/commands/shell.py @@ -3,8 +3,11 @@ Scrapy Shell See documentation in docs/topics/shell.rst """ +from argparse import Namespace from threading import Thread +from typing import List, Type +from scrapy import Spider from scrapy.commands import ScrapyCommand from scrapy.http import Request from scrapy.shell import Shell @@ -54,15 +57,16 @@ class Command(ScrapyCommand): """ pass - def run(self, args, opts): + def run(self, args: List[str], opts: Namespace) -> None: url = args[0] if args else None if url: # first argument may be a local file url = guess_scheme(url) + assert self.crawler_process spider_loader = self.crawler_process.spider_loader - spidercls = DefaultSpider + spidercls: Type[Spider] = DefaultSpider if opts.spider: spidercls = spider_loader.load(opts.spider) elif url: diff --git a/scrapy/http/response/text.py b/scrapy/http/response/text.py index f228e11c1..7fc54b5d3 100644 --- a/scrapy/http/response/text.py +++ b/scrapy/http/response/text.py @@ -4,10 +4,11 @@ discovering (through HTTP headers) to base Response class. See documentation in docs/topics/request-response.rst """ +from __future__ import annotations import json from contextlib import suppress -from typing import Generator, Optional, Tuple +from typing import TYPE_CHECKING, Any, Generator, Optional, Tuple from urllib.parse import urljoin import parsel @@ -25,6 +26,9 @@ from scrapy.http.response import Response from scrapy.utils.python import memoizemethod_noargs, to_unicode from scrapy.utils.response import get_base_url +if TYPE_CHECKING: + from scrapy.selector import Selector + _NONE = object() @@ -34,11 +38,11 @@ class TextResponse(Response): attributes: Tuple[str, ...] = Response.attributes + ("encoding",) - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any): self._encoding = kwargs.pop("encoding", None) - self._cached_benc = None + self._cached_benc: Optional[str] = None self._cached_ubody: Optional[str] = None - self._cached_selector = None + self._cached_selector: Optional[Selector] = None super().__init__(*args, **kwargs) def _set_url(self, url): diff --git a/scrapy/spiderloader.py b/scrapy/spiderloader.py index 13d6f9f87..f6bb93ddc 100644 --- a/scrapy/spiderloader.py +++ b/scrapy/spiderloader.py @@ -27,7 +27,7 @@ class SpiderLoader: self._found: DefaultDict[str, List[Tuple[str, str]]] = defaultdict(list) self._load_all_spiders() - def _check_name_duplicates(self): + def _check_name_duplicates(self) -> None: dupes = [] for name, locations in self._found.items(): dupes.extend( diff --git a/scrapy/utils/ossignal.py b/scrapy/utils/ossignal.py index f835a2221..2334ea792 100644 --- a/scrapy/utils/ossignal.py +++ b/scrapy/utils/ossignal.py @@ -1,7 +1,13 @@ import signal -from typing import Callable +from types import FrameType +from typing import Any, Callable, Dict, Optional, Union -signal_names = {} +# copy of _HANDLER from typeshed/stdlib/signal.pyi +SignalHandlerT = Union[ + Callable[[int, Optional[FrameType]], Any], int, signal.Handlers, None +] + +signal_names: Dict[int, str] = {} for signame in dir(signal): if signame.startswith("SIG") and not signame.startswith("SIG_"): signum = getattr(signal, signame) @@ -9,7 +15,9 @@ for signame in dir(signal): signal_names[signum] = signame -def install_shutdown_handlers(function: Callable, override_sigint: bool = True) -> None: +def install_shutdown_handlers( + function: SignalHandlerT, override_sigint: bool = True +) -> None: """Install the given function as a signal handler for all common shutdown signals (such as SIGINT, SIGTERM, etc). If override_sigint is ``False`` the SIGINT handler won't be install if there is already a handler in place diff --git a/scrapy/utils/spider.py b/scrapy/utils/spider.py index 3228eda49..704df8657 100644 --- a/scrapy/utils/spider.py +++ b/scrapy/utils/spider.py @@ -2,16 +2,19 @@ from __future__ import annotations import inspect import logging -from types import ModuleType +from types import CoroutineType, ModuleType from typing import ( TYPE_CHECKING, Any, - AsyncIterable, + AsyncGenerator, Generator, Iterable, + Literal, Optional, Type, + TypeVar, Union, + overload, ) from twisted.internet.defer import Deferred @@ -26,8 +29,26 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +_T = TypeVar("_T") -def iterate_spider_output(result: Any) -> Union[Iterable, AsyncIterable, Deferred]: + +# https://stackoverflow.com/questions/60222982 +@overload +def iterate_spider_output(result: AsyncGenerator) -> AsyncGenerator: # type: ignore[misc] + ... + + +@overload +def iterate_spider_output(result: CoroutineType) -> Deferred: + ... + + +@overload +def iterate_spider_output(result: _T) -> Iterable: + ... + + +def iterate_spider_output(result: Any) -> Union[Iterable, AsyncGenerator, Deferred]: if inspect.isasyncgen(result): return result if inspect.iscoroutine(result): @@ -55,6 +76,39 @@ def iter_spider_classes(module: ModuleType) -> Generator[Type[Spider], Any, None yield obj +@overload +def spidercls_for_request( + spider_loader: SpiderLoader, + request: Request, + default_spidercls: Type[Spider], + log_none: bool = ..., + log_multiple: bool = ..., +) -> Type[Spider]: + ... + + +@overload +def spidercls_for_request( + spider_loader: SpiderLoader, + request: Request, + default_spidercls: Literal[None], + log_none: bool = ..., + log_multiple: bool = ..., +) -> Optional[Type[Spider]]: + ... + + +@overload +def spidercls_for_request( + spider_loader: SpiderLoader, + request: Request, + *, + log_none: bool = ..., + log_multiple: bool = ..., +) -> Optional[Type[Spider]]: + ... + + def spidercls_for_request( spider_loader: SpiderLoader, request: Request,