Misc typing improvements. (#6384)

scrapy/core/http2/agent.py

@@ -119,7 +119,7 @@ class H2Agent:
             self._reactor, self._context_factory, connect_timeout, bind_address
         )
 
-    def get_endpoint(self, uri: URI):
+    def get_endpoint(self, uri: URI) -> HostnameEndpoint:
         return self.endpoint_factory.endpointForURI(uri)
 
     def get_key(self, uri: URI) -> Tuple:
@@ -161,7 +161,7 @@ class ScrapyProxyH2Agent(H2Agent):
         )
         self._proxy_uri = proxy_uri
 
-    def get_endpoint(self, uri: URI):
+    def get_endpoint(self, uri: URI) -> HostnameEndpoint:
         return self.endpoint_factory.endpointForURI(self._proxy_uri)
 
     def get_key(self, uri: URI) -> Tuple:

scrapy/core/http2/protocol.py

@@ -22,7 +22,11 @@ from h2.events import (
 from h2.exceptions import FrameTooLargeError, H2Error
 from twisted.internet.defer import Deferred
 from twisted.internet.error import TimeoutError
-from twisted.internet.interfaces import IHandshakeListener, IProtocolNegotiationFactory
+from twisted.internet.interfaces import (
+    IAddress,
+    IHandshakeListener,
+    IProtocolNegotiationFactory,
+)
 from twisted.internet.protocol import Factory, Protocol, connectionDone
 from twisted.internet.ssl import Certificate
 from twisted.protocols.policies import TimeoutMixin
@@ -431,7 +435,7 @@ class H2ClientFactory(Factory):
         self.settings = settings
         self.conn_lost_deferred = conn_lost_deferred
 
-    def buildProtocol(self, addr) -> H2ClientProtocol:
+    def buildProtocol(self, addr: IAddress) -> H2ClientProtocol:
         return H2ClientProtocol(self.uri, self.settings, self.conn_lost_deferred)
 
     def acceptableProtocols(self) -> List[bytes]:

scrapy/core/http2/stream.py

@@ -1,7 +1,7 @@
 import logging
 from enum import Enum
 from io import BytesIO
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 
 from h2.errors import ErrorCodes
 from h2.exceptions import H2Error, ProtocolError, StreamClosedError
@@ -142,7 +142,7 @@ class Stream:
             "headers": Headers({}),
         }
 
-        def _cancel(_) -> None:
+        def _cancel(_: Any) -> None:
             # Close this stream as gracefully as possible
             # If the associated request is initiated we reset this stream
             # else we directly call close() method

scrapy/downloadermiddlewares/httpcompression.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import warnings
 from itertools import chain
 from logging import getLogger
-from typing import TYPE_CHECKING, List, Optional, Union
+from typing import TYPE_CHECKING, List, Optional, Tuple, Union
 
 from scrapy import Request, Spider, signals
 from scrapy.crawler import Crawler
@@ -149,20 +149,24 @@ class HttpCompressionMiddleware:
 
         return response
 
-    def _handle_encoding(self, body, content_encoding, max_size):
+    def _handle_encoding(
+        self, body: bytes, content_encoding: List[bytes], max_size: int
+    ) -> Tuple[bytes, List[bytes]]:
         to_decode, to_keep = self._split_encodings(content_encoding)
         for encoding in to_decode:
             body = self._decode(body, encoding, max_size)
         return body, to_keep
 
-    def _split_encodings(self, content_encoding):
-        to_keep = [
+    def _split_encodings(
+        self, content_encoding: List[bytes]
+    ) -> Tuple[List[bytes], List[bytes]]:
+        to_keep: List[bytes] = [
             encoding.strip().lower()
             for encoding in chain.from_iterable(
                 encodings.split(b",") for encodings in content_encoding
             )
         ]
-        to_decode = []
+        to_decode: List[bytes] = []
         while to_keep:
             encoding = to_keep.pop()
             if encoding not in ACCEPTED_ENCODINGS:

scrapy/downloadermiddlewares/offsite.py

@@ -1,33 +1,43 @@
+from __future__ import annotations
+
 import logging
 import re
 import warnings
+from typing import TYPE_CHECKING, Set
 
-from scrapy import signals
+from scrapy import Request, Spider, signals
+from scrapy.crawler import Crawler
 from scrapy.exceptions import IgnoreRequest
+from scrapy.statscollectors import StatsCollector
 from scrapy.utils.httpobj import urlparse_cached
 
+if TYPE_CHECKING:
+    # typing.Self requires Python 3.11
+    from typing_extensions import Self
+
 logger = logging.getLogger(__name__)
 
 
 class OffsiteMiddleware:
     @classmethod
-    def from_crawler(cls, crawler):
+    def from_crawler(cls, crawler: Crawler) -> Self:
+        assert crawler.stats
         o = cls(crawler.stats)
         crawler.signals.connect(o.spider_opened, signal=signals.spider_opened)
         crawler.signals.connect(o.request_scheduled, signal=signals.request_scheduled)
         return o
 
-    def __init__(self, stats):
+    def __init__(self, stats: StatsCollector):
         self.stats = stats
-        self.domains_seen = set()
+        self.domains_seen: Set[str] = set()
 
-    def spider_opened(self, spider):
-        self.host_regex = self.get_host_regex(spider)
+    def spider_opened(self, spider: Spider) -> None:
+        self.host_regex: re.Pattern[str] = self.get_host_regex(spider)
 
-    def request_scheduled(self, request, spider):
+    def request_scheduled(self, request: Request, spider: Spider) -> None:
         self.process_request(request, spider)
 
-    def process_request(self, request, spider):
+    def process_request(self, request: Request, spider: Spider) -> None:
         if request.dont_filter or self.should_follow(request, spider):
             return None
         domain = urlparse_cached(request).hostname
@@ -42,13 +52,13 @@ class OffsiteMiddleware:
         self.stats.inc_value("offsite/filtered", spider=spider)
         raise IgnoreRequest
 
-    def should_follow(self, request, spider):
+    def should_follow(self, request: Request, spider: Spider) -> bool:
         regex = self.host_regex
         # hostname can be None for wrong urls (like javascript links)
         host = urlparse_cached(request).hostname or ""
         return bool(regex.search(host))
 
-    def get_host_regex(self, spider):
+    def get_host_regex(self, spider: Spider) -> re.Pattern[str]:
         """Override this method to implement a different offsite policy"""
         allowed_domains = getattr(spider, "allowed_domains", None)
         if not allowed_domains:

scrapy/loader/__init__.py

@@ -4,8 +4,11 @@ Item Loader
 See documentation in docs/topics/loaders.rst
 """
 
+from typing import Any, Optional
+
 import itemloaders
 
+from scrapy.http import TextResponse
 from scrapy.item import Item
 from scrapy.selector import Selector
 
@@ -82,7 +85,14 @@ class ItemLoader(itemloaders.ItemLoader):
     default_item_class: type = Item
     default_selector_class = Selector
 
-    def __init__(self, item=None, selector=None, response=None, parent=None, **context):
+    def __init__(
+        self,
+        item: Any = None,
+        selector: Optional[Selector] = None,
+        response: Optional[TextResponse] = None,
+        parent: Optional[itemloaders.ItemLoader] = None,
+        **context: Any
+    ):
         if selector is None and response is not None:
             try:
                 selector = self.default_selector_class(response)

scrapy/utils/benchserver.py

@@ -1,21 +1,23 @@
 import random
+from typing import Any
 from urllib.parse import urlencode
 
 from twisted.web.resource import Resource
-from twisted.web.server import Site
+from twisted.web.server import Request, Site
 
 
 class Root(Resource):
     isLeaf = True
 
-    def getChild(self, name, request):
+    def getChild(self, name: str, request: Request) -> Resource:
         return self
 
-    def render(self, request):
+    def render(self, request: Request) -> bytes:
         total = _getarg(request, b"total", 100, int)
         show = _getarg(request, b"show", 10, int)
         nlist = [random.randint(1, total) for _ in range(show)]  # nosec
         request.write(b"<html><head></head><body>")
+        assert request.args is not None
         args = request.args.copy()
         for nl in nlist:
             args["n"] = nl
@@ -27,7 +29,7 @@ class Root(Resource):
         return b""
 
 
-def _getarg(request, name, default=None, type=str):
+def _getarg(request, name: bytes, default: Any = None, type=str):
     return type(request.args[name][0]) if name in request.args else default
 
 
@@ -38,7 +40,7 @@ if __name__ == "__main__":
     factory = Site(root)
     httpPort = reactor.listenTCP(8998, Site(root))
 
-    def _print_listening():
+    def _print_listening() -> None:
         httpHost = httpPort.getHost()
         print(f"Bench server at http://{httpHost.host}:{httpHost.port}")
 

scrapy/utils/curl.py

@@ -2,13 +2,20 @@ import argparse
 import warnings
 from http.cookies import SimpleCookie
 from shlex import split
+from typing import Any, Dict, List, NoReturn, Optional, Sequence, Tuple, Union
 from urllib.parse import urlparse
 
 from w3lib.http import basic_auth_header
 
 
 class DataAction(argparse.Action):
-    def __call__(self, parser, namespace, values, option_string=None):
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: Union[str, Sequence[Any], None],
+        option_string: Optional[str] = None,
+    ) -> None:
         value = str(values)
         if value.startswith("$"):
             value = value[1:]
@@ -16,7 +23,7 @@ class DataAction(argparse.Action):
 
 
 class CurlParser(argparse.ArgumentParser):
-    def error(self, message):
+    def error(self, message: str) -> NoReturn:
         error_msg = f"There was an error parsing the curl command: {message}"
         raise ValueError(error_msg)
 
@@ -42,9 +49,11 @@ for argument in safe_to_ignore_arguments:
     curl_parser.add_argument(*argument, action="store_true")
 
 
-def _parse_headers_and_cookies(parsed_args):
-    headers = []
-    cookies = {}
+def _parse_headers_and_cookies(
+    parsed_args: argparse.Namespace,
+) -> Tuple[List[Tuple[str, bytes]], Dict[str, str]]:
+    headers: List[Tuple[str, bytes]] = []
+    cookies: Dict[str, str] = {}
     for header in parsed_args.headers or ():
         name, val = header.split(":", 1)
         name = name.strip()
@@ -64,7 +73,7 @@ def _parse_headers_and_cookies(parsed_args):
 
 def curl_to_request_kwargs(
     curl_command: str, ignore_unknown_options: bool = True
-) -> dict:
+) -> Dict[str, Any]:
     """Convert a cURL command syntax to Request kwargs.
 
     :param str curl_command: string containing the curl command
@@ -98,7 +107,7 @@ def curl_to_request_kwargs(
 
     method = parsed_args.method or "GET"
 
-    result = {"method": method.upper(), "url": url}
+    result: Dict[str, Any] = {"method": method.upper(), "url": url}
 
     headers, cookies = _parse_headers_and_cookies(parsed_args)
 

scrapy/utils/datatypes.py

@@ -110,7 +110,7 @@ class CaseInsensitiveDict(collections.UserDict):
     as keys and allows case-insensitive lookups.
     """
 
-    def __init__(self, *args, **kwargs) -> None:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         self._keys: dict = {}
         super().__init__(*args, **kwargs)
 

scrapy/utils/request.py

@@ -138,7 +138,7 @@ class RequestFingerprinter:
     """
 
     @classmethod
-    def from_crawler(cls, crawler) -> Self:
+    def from_crawler(cls, crawler: Crawler) -> Self:
         return cls(crawler)
 
     def __init__(self, crawler: Optional[Crawler] = None):

scrapy/utils/response.py

@@ -58,7 +58,7 @@ def response_status_message(status: Union[bytes, float, int, str]) -> str:
     return f"{status_int} {to_unicode(message)}"
 
 
-def _remove_html_comments(body):
+def _remove_html_comments(body: bytes) -> bytes:
     start = body.find(b"<!--")
     while start != -1:
         end = body.find(b"-->", start + 1)

scrapy/utils/testsite.py

@@ -15,12 +15,12 @@ class SiteTest:
         super().tearDown()
         self.site.stopListening()
 
-    def url(self, path):
+    def url(self, path: str) -> str:
         return urljoin(self.baseurl, path)
 
 
 class NoMetaRefreshRedirect(util.Redirect):
-    def render(self, request):
+    def render(self, request: server.Request) -> bytes:
         content = util.Redirect.render(self, request)
         return content.replace(
             b'http-equiv="refresh"', b'http-no-equiv="do-not-refresh-me"'