Add typing to scrapy/utils/datatypes.py.

scrapy/resolver.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from twisted.internet import defer
 from twisted.internet.base import ThreadedResolver
 from twisted.internet.interfaces import (
@@ -11,7 +13,7 @@ from zope.interface.declarations import implementer, provider
 from scrapy.utils.datatypes import LocalCache
 
 # TODO: cache misses
-dnscache = LocalCache(10000)
+dnscache: LocalCache[str, Any] = LocalCache(10000)
 
 
 @implementer(IResolverSimple)
@@ -36,7 +38,7 @@ class CachingThreadedResolver(ThreadedResolver):
     def install_on_reactor(self):
         self.reactor.installResolver(self)
 
-    def getHostByName(self, name, timeout=None):
+    def getHostByName(self, name: str, timeout=None):
         if name in dnscache:
             return defer.succeed(dnscache[name])
         # in Twisted<=16.6, getHostByName() is always called with
@@ -110,7 +112,7 @@ class CachingHostnameResolver:
     def resolveHostName(
         self,
         resolutionReceiver,
-        hostName,
+        hostName: str,
         portNumber=0,
         addressTypes=None,
         transportSemantics="TCP",

scrapy/utils/datatypes.py

@@ -8,6 +8,10 @@ This module must not depend on any module outside the Standard Library.
 import collections
 import weakref
 from collections.abc import Mapping
+from typing import Any, Optional, Sequence, TypeVar
+
+_KT = TypeVar("_KT")
+_VT = TypeVar("_VT")
 
 
 class CaselessDict(dict):
@@ -64,24 +68,24 @@ class CaselessDict(dict):
         return dict.pop(self, self.normkey(key), *args)
 
 
-class LocalCache(collections.OrderedDict):
+class LocalCache(collections.OrderedDict[_KT, _VT]):
     """Dictionary with a finite number of keys.
 
     Older items expires first.
     """
 
-    def __init__(self, limit=None):
+    def __init__(self, limit: Optional[int] = None):
         super().__init__()
-        self.limit = limit
+        self.limit: Optional[int] = limit
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: _KT, value: _VT) -> None:
         if self.limit:
             while len(self) >= self.limit:
                 self.popitem(last=False)
         super().__setitem__(key, value)
 
 
-class LocalWeakReferencedCache(weakref.WeakKeyDictionary):
+class LocalWeakReferencedCache(weakref.WeakKeyDictionary[_KT, _VT]):
     """
     A weakref.WeakKeyDictionary implementation that uses LocalCache as its
     underlying data structure, making it ordered and capable of being size-limited.
@@ -93,17 +97,17 @@ class LocalWeakReferencedCache(weakref.WeakKeyDictionary):
     it cannot be instantiated with an initial dictionary.
     """
 
-    def __init__(self, limit=None):
+    def __init__(self, limit: Optional[int] = None):
         super().__init__()
-        self.data = LocalCache(limit=limit)
+        self.data: LocalCache = LocalCache(limit=limit)
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: _KT, value: _VT) -> None:
         try:
             super().__setitem__(key, value)
         except TypeError:
             pass  # key is not weak-referenceable, skip caching
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: _KT) -> Optional[_VT]:  # type: ignore[override]
         try:
             return super().__getitem__(key)
         except (TypeError, KeyError):
@@ -113,8 +117,8 @@ class LocalWeakReferencedCache(weakref.WeakKeyDictionary):
 class SequenceExclude:
     """Object to test if an item is NOT within some sequence."""
 
-    def __init__(self, seq):
-        self.seq = seq
+    def __init__(self, seq: Sequence):
+        self.seq: Sequence = seq
 
-    def __contains__(self, item):
+    def __contains__(self, item: Any) -> bool:
         return item not in self.seq

scrapy/utils/misc.py

@@ -214,16 +214,18 @@ def walk_callable(node):
         yield node
 
 
-_generator_callbacks_cache = LocalWeakReferencedCache(limit=128)
+_generator_callbacks_cache: LocalWeakReferencedCache[
+    Callable, bool
+] = LocalWeakReferencedCache(limit=128)
 
 
-def is_generator_with_return_value(callable):
+def is_generator_with_return_value(callable: Callable) -> bool:
     """
     Returns True if a callable is a generator function which includes a
     'return' statement with a value different than None, False otherwise
     """
     if callable in _generator_callbacks_cache:
-        return _generator_callbacks_cache[callable]
+        return bool(_generator_callbacks_cache[callable])
 
     def returns_none(return_node):
         value = return_node.value
@@ -248,10 +250,10 @@ def is_generator_with_return_value(callable):
         for node in walk_callable(tree):
             if isinstance(node, ast.Return) and not returns_none(node):
                 _generator_callbacks_cache[callable] = True
-                return _generator_callbacks_cache[callable]
+                return bool(_generator_callbacks_cache[callable])
 
     _generator_callbacks_cache[callable] = False
-    return _generator_callbacks_cache[callable]
+    return bool(_generator_callbacks_cache[callable])
 
 
 def warn_on_generator_with_return_value(spider: "Spider", callable: Callable) -> None: