Full typing for scrapy/resolver.py.

scrapy/resolver.py

@@ -1,8 +1,12 @@
-from typing import Any
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, List, Optional, Sequence, Type
 
 from twisted.internet import defer
-from twisted.internet.base import ThreadedResolver
+from twisted.internet.base import ReactorBase, ThreadedResolver
+from twisted.internet.defer import Deferred
 from twisted.internet.interfaces import (
+    IAddress,
     IHostnameResolver,
     IHostResolution,
     IResolutionReceiver,
@@ -12,6 +16,12 @@ from zope.interface.declarations import implementer, provider
 
 from scrapy.utils.datatypes import LocalCache
 
+if TYPE_CHECKING:
+    # typing.Self requires Python 3.11
+    from typing_extensions import Self
+
+    from scrapy.crawler import Crawler
+
 # TODO: cache misses
 dnscache: LocalCache[str, Any] = LocalCache(10000)
 
@@ -22,65 +32,66 @@ class CachingThreadedResolver(ThreadedResolver):
     Default caching resolver. IPv4 only, supports setting a timeout value for DNS requests.
     """
 
-    def __init__(self, reactor, cache_size, timeout):
+    def __init__(self, reactor: ReactorBase, cache_size: int, timeout: float):
         super().__init__(reactor)
         dnscache.limit = cache_size
         self.timeout = timeout
 
     @classmethod
-    def from_crawler(cls, crawler, reactor):
+    def from_crawler(cls, crawler: Crawler, reactor: ReactorBase) -> Self:
         if crawler.settings.getbool("DNSCACHE_ENABLED"):
             cache_size = crawler.settings.getint("DNSCACHE_SIZE")
         else:
             cache_size = 0
         return cls(reactor, cache_size, crawler.settings.getfloat("DNS_TIMEOUT"))
 
-    def install_on_reactor(self):
+    def install_on_reactor(self) -> None:
         self.reactor.installResolver(self)
 
-    def getHostByName(self, name: str, timeout=None):
+    def getHostByName(self, name: str, timeout: Sequence[int] = ()) -> Deferred[str]:
         if name in dnscache:
             return defer.succeed(dnscache[name])
         # in Twisted<=16.6, getHostByName() is always called with
         # a default timeout of 60s (actually passed as (1, 3, 11, 45) tuple),
         # so the input argument above is simply overridden
         # to enforce Scrapy's DNS_TIMEOUT setting's value
-        timeout = (self.timeout,)
+        # The timeout arg is typed as Sequence[int] but supports floats.
+        timeout = (self.timeout,)  # type: ignore[assignment]
         d = super().getHostByName(name, timeout)
         if dnscache.limit:
             d.addCallback(self._cache_result, name)
         return d
 
-    def _cache_result(self, result, name):
+    def _cache_result(self, result: Any, name: str) -> Any:
         dnscache[name] = result
         return result
 
 
 @implementer(IHostResolution)
 class HostResolution:
-    def __init__(self, name):
-        self.name = name
+    def __init__(self, name: str):
+        self.name: str = name
 
-    def cancel(self):
+    def cancel(self) -> None:
         raise NotImplementedError()
 
 
 @provider(IResolutionReceiver)
 class _CachingResolutionReceiver:
-    def __init__(self, resolutionReceiver, hostName):
-        self.resolutionReceiver = resolutionReceiver
-        self.hostName = hostName
-        self.addresses = []
+    def __init__(self, resolutionReceiver: IResolutionReceiver, hostName: str):
+        self.resolutionReceiver: IResolutionReceiver = resolutionReceiver
+        self.hostName: str = hostName
+        self.addresses: List[IAddress] = []
 
-    def resolutionBegan(self, resolution):
+    def resolutionBegan(self, resolution: IHostResolution) -> None:
         self.resolutionReceiver.resolutionBegan(resolution)
         self.resolution = resolution
 
-    def addressResolved(self, address):
+    def addressResolved(self, address: IAddress) -> None:
         self.resolutionReceiver.addressResolved(address)
         self.addresses.append(address)
 
-    def resolutionComplete(self):
+    def resolutionComplete(self) -> None:
         self.resolutionReceiver.resolutionComplete()
         if self.addresses:
             dnscache[self.hostName] = self.addresses
@@ -93,30 +104,30 @@ class CachingHostnameResolver:
     does not support setting a timeout value for DNS requests.
     """
 
-    def __init__(self, reactor, cache_size):
-        self.reactor = reactor
-        self.original_resolver = reactor.nameResolver
+    def __init__(self, reactor: ReactorBase, cache_size: int):
+        self.reactor: ReactorBase = reactor
+        self.original_resolver: IHostnameResolver = reactor.nameResolver
         dnscache.limit = cache_size
 
     @classmethod
-    def from_crawler(cls, crawler, reactor):
+    def from_crawler(cls, crawler: Crawler, reactor: ReactorBase) -> Self:
         if crawler.settings.getbool("DNSCACHE_ENABLED"):
             cache_size = crawler.settings.getint("DNSCACHE_SIZE")
         else:
             cache_size = 0
         return cls(reactor, cache_size)
 
-    def install_on_reactor(self):
+    def install_on_reactor(self) -> None:
         self.reactor.installNameResolver(self)
 
     def resolveHostName(
         self,
-        resolutionReceiver,
+        resolutionReceiver: IResolutionReceiver,
         hostName: str,
-        portNumber=0,
-        addressTypes=None,
-        transportSemantics="TCP",
-    ):
+        portNumber: int = 0,
+        addressTypes: Optional[Sequence[Type[IAddress]]] = None,
+        transportSemantics: str = "TCP",
+    ) -> IHostResolution:
         try:
             addresses = dnscache[hostName]
         except KeyError: