More types for synapse.util, part 1

changelog.d/10888.misc

@@ -0,0 +1 @@
+Improve type hinting in `synapse.util`.

mypy.ini

@@ -100,9 +100,30 @@ disallow_untyped_defs = True
 [mypy-synapse.util.batching_queue]
 disallow_untyped_defs = True
 
+[mypy-synapse.util.caches.cached_call]
+disallow_untyped_defs = True
+
+[mypy-synapse.util.caches.deferred_cache]
+disallow_untyped_defs = True
+
 [mypy-synapse.util.caches.dictionary_cache]
 disallow_untyped_defs = True
 
+[mypy-synapse.util.caches.lrucache]
+disallow_untyped_defs = True
+
+[mypy-synapse.util.caches.response_cache]
+disallow_untyped_defs = True
+
+[mypy-synapse.util.caches.stream_change_cache]
+disallow_untyped_defs = True
+
+[mypy-synapse.util.caches.ttl_cache]
+disallow_untyped_defs = True
+
+[mypy-synapse.util.daemonize]
+disallow_untyped_defs = True
+
 [mypy-synapse.util.file_consumer]
 disallow_untyped_defs = True
 
@@ -139,6 +160,9 @@ disallow_untyped_defs = True
 [mypy-synapse.util.msisdn]
 disallow_untyped_defs = True
 
+[mypy-synapse.util.patch_inline_callbacks]
+disallow_untyped_defs = True
+
 [mypy-synapse.util.ratelimitutils]
 disallow_untyped_defs = True
 
@@ -160,6 +184,9 @@ disallow_untyped_defs = True
 [mypy-synapse.util.wheel_timer]
 disallow_untyped_defs = True
 
+[mypy-synapse.util.versionstring]
+disallow_untyped_defs = True
+
 [mypy-pymacaroons.*]
 ignore_missing_imports = True
 

synapse/util/caches/cached_call.py

@@ -85,7 +85,7 @@ async def get(self) -> TV:
             # result in the deferred, since `awaiting` a deferred destroys its result.
             # (Also, if it's a Failure, GCing the deferred would log a critical error
             # about unhandled Failures)
-            def got_result(r):
+            def got_result(r: Union[TV, Failure]) -> None:
                 self._result = r
 
             self._deferred.addBoth(got_result)

synapse/util/caches/deferred_cache.py

@@ -31,6 +31,7 @@
 
 from twisted.internet import defer
 from twisted.python import failure
+from twisted.python.failure import Failure
 
 from synapse.util.async_helpers import ObservableDeferred
 from synapse.util.caches.lrucache import LruCache
@@ -110,7 +111,7 @@ def metrics_cb() -> None:
         self.thread: Optional[threading.Thread] = None
 
     @property
-    def max_entries(self):
+    def max_entries(self) -> int:
         return self.cache.max_size
 
     def check_thread(self) -> None:
@@ -256,7 +257,7 @@ def compare_and_pop() -> bool:
 
             return False
 
-        def cb(result) -> None:
+        def cb(result: VT) -> None:
             if compare_and_pop():
                 self.cache.set(key, result, entry.callbacks)
             else:
@@ -268,7 +269,7 @@ def cb(result) -> None:
                 # not have been. Either way, let's double-check now.
                 entry.invalidate()
 
-        def eb(_fail) -> None:
+        def eb(_fail: Failure) -> None:
             compare_and_pop()
             entry.invalidate()
 
@@ -282,11 +283,11 @@ def eb(_fail) -> None:
 
     def prefill(
         self, key: KT, value: VT, callback: Optional[Callable[[], None]] = None
-    ):
+    ) -> None:
         callbacks = [callback] if callback else []
         self.cache.set(key, value, callbacks=callbacks)
 
-    def invalidate(self, key):
+    def invalidate(self, key: KT) -> None:
         """Delete a key, or tree of entries
 
         If the cache is backed by a regular dict, then "key" must be of

synapse/util/caches/lrucache.py

@@ -21,6 +21,7 @@
     Any,
     Callable,
     Collection,
+    Dict,
     Generic,
     Iterable,
     List,
@@ -52,7 +53,7 @@
 try:
     from pympler.asizeof import Asizer
 
-    def _get_size_of(val: Any, *, recurse=True) -> int:
+    def _get_size_of(val: Any, *, recurse: bool = True) -> int:
         """Get an estimate of the size in bytes of the object.
 
         Args:
@@ -71,7 +72,7 @@ def _get_size_of(val: Any, *, recurse=True) -> int:
 
 except ImportError:
 
-    def _get_size_of(val: Any, *, recurse=True) -> int:
+    def _get_size_of(val: Any, *, recurse: bool = True) -> int:
         return 0
 
 
@@ -86,7 +87,8 @@ def _get_size_of(val: Any, *, recurse=True) -> int:
 T = TypeVar("T")
 
 
-def enumerate_leaves(node, depth):
+# TODO I think this is this unused?
+def enumerate_leaves(node: Dict, depth: int) -> Any:
     if depth == 0:
         yield node
     else:
@@ -102,7 +104,7 @@ class _TimedListNode(ListNode[P]):
 
     __slots__ = ["last_access_ts_secs"]
 
-    def update_last_access(self, clock: Clock):
+    def update_last_access(self, clock: Clock) -> None:
         self.last_access_ts_secs = int(clock.time())
 
 
@@ -115,7 +117,7 @@ def update_last_access(self, clock: Clock):
 
 
 @wrap_as_background_process("LruCache._expire_old_entries")
-async def _expire_old_entries(clock: Clock, expiry_seconds: int):
+async def _expire_old_entries(clock: Clock, expiry_seconds: int) -> None:
     """Walks the global cache list to find cache entries that haven't been
     accessed in the given number of seconds.
     """
@@ -163,7 +165,7 @@ async def _expire_old_entries(clock: Clock, expiry_seconds: int):
     logger.info("Dropped %d items from caches", i)
 
 
-def setup_expire_lru_cache_entries(hs: "HomeServer"):
+def setup_expire_lru_cache_entries(hs: "HomeServer") -> None:
     """Start a background job that expires all cache entries if they have not
     been accessed for the given number of seconds.
     """
@@ -183,7 +185,7 @@ def setup_expire_lru_cache_entries(hs: "HomeServer"):
     )
 
 
-class _Node:
+class _Node(Generic[KT, VT]):
     __slots__ = [
         "_list_node",
         "_global_list_node",
@@ -197,8 +199,8 @@ class _Node:
     def __init__(
         self,
         root: "ListNode[_Node]",
-        key,
-        value,
+        key: KT,
+        value: VT,
         cache: "weakref.ReferenceType[LruCache]",
         clock: Clock,
         callbacks: Collection[Callable[[], None]] = (),
@@ -407,7 +409,7 @@ def evict() -> None:
 
         def synchronized(f: FT) -> FT:
             @wraps(f)
-            def inner(*args, **kwargs):
+            def inner(*args: Any, **kwargs: Any) -> Any:
                 with lock:
                     return f(*args, **kwargs)
 
@@ -416,17 +418,19 @@ def inner(*args, **kwargs):
         cached_cache_len = [0]
         if size_callback is not None:
 
-            def cache_len():
+            def cache_len() -> int:
                 return cached_cache_len[0]
 
         else:
 
-            def cache_len():
+            def cache_len() -> int:
                 return len(cache)
 
         self.len = synchronized(cache_len)
 
-        def add_node(key, value, callbacks: Collection[Callable[[], None]] = ()):
+        def add_node(
+            key: KT, value: VT, callbacks: Collection[Callable[[], None]] = ()
+        ) -> None:
             node = _Node(list_root, key, value, weak_ref_to_self, real_clock, callbacks)
             cache[key] = node
 
@@ -436,7 +440,7 @@ def add_node(key, value, callbacks: Collection[Callable[[], None]] = ()):
             if caches.TRACK_MEMORY_USAGE and metrics:
                 metrics.inc_memory_usage(node.memory)
 
-        def move_node_to_front(node: _Node):
+        def move_node_to_front(node: _Node) -> None:
             node.move_to_front(real_clock, list_root)
 
         def delete_node(node: _Node) -> int:
@@ -478,7 +482,7 @@ def cache_get(
             default: Optional[T] = None,
             callbacks: Collection[Callable[[], None]] = (),
             update_metrics: bool = True,
-        ):
+        ) -> Union[None, T, VT]:
             node = cache.get(key, None)
             if node is not None:
                 move_node_to_front(node)
@@ -492,7 +496,9 @@ def cache_get(
                 return default
 
         @synchronized
-        def cache_set(key: KT, value: VT, callbacks: Iterable[Callable[[], None]] = ()):
+        def cache_set(
+            key: KT, value: VT, callbacks: Iterable[Callable[[], None]] = ()
+        ) -> None:
             node = cache.get(key, None)
             if node is not None:
                 # We sometimes store large objects, e.g. dicts, which cause
@@ -537,7 +543,7 @@ def cache_pop(key: KT, default: T) -> Union[T, VT]:
             ...
 
         @synchronized
-        def cache_pop(key: KT, default: Optional[T] = None):
+        def cache_pop(key: KT, default: Optional[T] = None) -> Union[None, T, VT]:
             node = cache.get(key, None)
             if node:
                 delete_node(node)
@@ -602,25 +608,25 @@ def cache_contains(key: KT) -> bool:
         self.contains = cache_contains
         self.clear = cache_clear
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: KT) -> VT:
         result = self.get(key, self.sentinel)
         if result is self.sentinel:
             raise KeyError()
         else:
-            return result
+            return cast(VT, result)
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: KT, value: VT) -> None:
         self.set(key, value)
 
-    def __delitem__(self, key, value):
+    def __delitem__(self, key: KT, value: VT) -> None:
         result = self.pop(key, self.sentinel)
         if result is self.sentinel:
             raise KeyError()
 
-    def __len__(self):
+    def __len__(self) -> int:
         return self.len()
 
-    def __contains__(self, key):
+    def __contains__(self, key: KT) -> bool:
         return self.contains(key)
 
     def set_cache_factor(self, factor: float) -> bool:

synapse/util/caches/response_cache.py

@@ -104,8 +104,8 @@ def get(self, key: KV) -> Optional[defer.Deferred]:
             return None
 
     def _set(
-        self, context: ResponseCacheContext[KV], deferred: defer.Deferred
-    ) -> defer.Deferred:
+        self, context: ResponseCacheContext[KV], deferred: defer.Deferred[RV]
+    ) -> defer.Deferred[RV]:
         """Set the entry for the given key to the given deferred.
 
         *deferred* should run its callbacks in the sentinel logcontext (ie,
@@ -126,7 +126,7 @@ def _set(
         key = context.cache_key
         self.pending_result_cache[key] = result
 
-        def on_complete(r):
+        def on_complete(r: RV) -> RV:
             # if this cache has a non-zero timeout, and the callback has not cleared
             # the should_cache bit, we leave it in the cache for now and schedule
             # its removal later.

synapse/util/caches/stream_change_cache.py

@@ -40,10 +40,10 @@ def __init__(
         self,
         name: str,
         current_stream_pos: int,
-        max_size=10000,
+        max_size: int = 10000,
         prefilled_cache: Optional[Mapping[EntityType, int]] = None,
-    ):
-        self._original_max_size = max_size
+    ) -> None:
+        self._original_max_size: int = max_size
         self._max_size = math.floor(max_size)
         self._entity_to_key: Dict[EntityType, int] = {}
 

synapse/util/caches/ttlcache.py

@@ -159,12 +159,12 @@ def expire(self) -> None:
             del self._expiry_list[0]
 
 
-@attr.s(frozen=True, slots=True)
-class _CacheEntry:
+@attr.s(frozen=True, slots=True, auto_attribs=True)
+class _CacheEntry(Generic[KT, VT]):
     """TTLCache entry"""
 
     # expiry_time is the first attribute, so that entries are sorted by expiry.
-    expiry_time = attr.ib(type=float)
-    ttl = attr.ib(type=float)
-    key = attr.ib()
-    value = attr.ib()
+    expiry_time: float
+    ttl: float
+    key: KT
+    value: VT

synapse/util/daemonize.py

@@ -19,6 +19,8 @@
 import os
 import signal
 import sys
+from types import FrameType, TracebackType
+from typing import NoReturn, Type
 
 
 def daemonize_process(pid_file: str, logger: logging.Logger, chdir: str = "/") -> None:
@@ -97,7 +99,9 @@ def daemonize_process(pid_file: str, logger: logging.Logger, chdir: str = "/") -
     # (we don't normally expect reactor.run to raise any exceptions, but this will
     # also catch any other uncaught exceptions before we get that far.)
 
-    def excepthook(type_, value, traceback):
+    def excepthook(
+        type_: Type[BaseException], value: BaseException, traceback: TracebackType
+    ) -> None:
         logger.critical("Unhanded exception", exc_info=(type_, value, traceback))
 
     sys.excepthook = excepthook
@@ -119,7 +123,7 @@ def excepthook(type_, value, traceback):
         sys.exit(1)
 
     # write a log line on SIGTERM.
-    def sigterm(signum, frame):
+    def sigterm(signum: signal.Signals, frame: FrameType) -> NoReturn:
         logger.warning("Caught signal %s. Stopping daemon." % signum)
         sys.exit(0)