Add types to synapse.util.

changelog.d/10601.misc

@@ -0,0 +1 @@
+Add type annotations to complete the synapse.util package.

mypy.ini

@@ -70,17 +70,7 @@ files =
   synapse/storage/util,
   synapse/streams,
   synapse/types.py,
-  synapse/util/async_helpers.py,
-  synapse/util/caches,
-  synapse/util/daemonize.py,
-  synapse/util/hash.py,
-  synapse/util/iterutils.py,
-  synapse/util/linked_list.py,
-  synapse/util/metrics.py,
-  synapse/util/macaroons.py,
-  synapse/util/module_loader.py,
-  synapse/util/msisdn.py,
-  synapse/util/stringutils.py,
+  synapse/util,
   synapse/visibility.py,
   tests/replication,
   tests/test_event_auth.py,

synapse/util/__init__.py

@@ -79,7 +79,7 @@ class Clock:
 
     @defer.inlineCallbacks
     def sleep(self, seconds):
-        d = defer.Deferred()
+        d: defer.Deferred = defer.Deferred()
         with context.PreserveLoggingContext():
             self._reactor.callLater(seconds, d.callback, seconds)
             res = yield d

synapse/util/batching_queue.py

@@ -122,7 +122,7 @@ async def add_to_queue(self, value: V, key: Hashable = ()) -> R:
 
         # First we create a defer and add it and the value to the list of
         # pending items.
-        d = defer.Deferred()
+        d: defer.Deferred = defer.Deferred()
         self._next_values.setdefault(key, []).append((value, d))
 
         # If we're not currently processing the key fire off a background

synapse/util/file_consumer.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import queue
+from typing import Optional
 
 from twisted.internet import threads
 
@@ -51,15 +52,15 @@ def __init__(self, file_obj, reactor):
 
         # Queue of slices of bytes to be written. When producer calls
         # unregister a final None is sent.
-        self._bytes_queue = queue.Queue()
+        self._bytes_queue: queue.Queue[Optional[bytes]] = queue.Queue()
 
         # Deferred that is resolved when finished writing
         self._finished_deferred = None
 
         # If the _writer thread throws an exception it gets stored here.
         self._write_exception = None
 
-    def registerProducer(self, producer, streaming):
+    def registerProducer(self, producer, streaming) -> None:
         """Part of IConsumer interface
 
         Args:
@@ -81,17 +82,19 @@ def registerProducer(self, producer, streaming):
         if not streaming:
             self._producer.resumeProducing()
 
-    def unregisterProducer(self):
+    def unregisterProducer(self) -> None:
         """Part of IProducer interface"""
         self._producer = None
+        assert self._finished_deferred is not None
         if not self._finished_deferred.called:
             self._bytes_queue.put_nowait(None)
 
-    def write(self, bytes):
+    def write(self, bytes) -> None:
         """Part of IProducer interface"""
         if self._write_exception:
             raise self._write_exception
 
+        assert self._finished_deferred is not None
         if self._finished_deferred.called:
             raise Exception("consumer has closed")
 
@@ -101,9 +104,10 @@ def write(self, bytes):
         # then we pause the producer.
         if self.streaming and self._bytes_queue.qsize() >= self._PAUSE_ON_QUEUE_SIZE:
             self._paused_producer = True
+            assert self._producer is not None
             self._producer.pauseProducing()
 
-    def _writer(self):
+    def _writer(self) -> None:
         """This is run in a background thread to write to the file."""
         try:
             while self._producer or not self._bytes_queue.empty():
@@ -134,7 +138,7 @@ def wait(self):
         """Returns a deferred that resolves when finished writing to file"""
         return make_deferred_yieldable(self._finished_deferred)
 
-    def _resume_paused_producer(self):
+    def _resume_paused_producer(self) -> None:
         """Gets called if we should resume producing after being paused"""
         if self._paused_producer and self._producer:
             self._paused_producer = False

synapse/util/manhole.py

@@ -79,7 +79,10 @@ def manhole(username, password, globals):
     checker = checkers.InMemoryUsernamePasswordDatabaseDontUse(**{username: password})
 
     rlm = manhole_ssh.TerminalRealm()
-    rlm.chainedProtocolFactory = lambda: insults.ServerProtocol(
+    # mypy ignored here because:
+    # - can't deduce types of lambdas
+    # - variable is Type[ServerProtocol], expr is Callable[[], ServerProtocol]
+    rlm.chainedProtocolFactory = lambda: insults.ServerProtocol(  # type: ignore[misc,assignment]
         SynapseManhole, dict(globals, __name__="__console__")
     )
 
@@ -110,6 +113,7 @@ def showsyntaxerror(self, filename=None):
         any syntax errors to be sent to the terminal, rather than sentry.
         """
         type, value, tb = sys.exc_info()
+        assert value is not None
         sys.last_type = type
         sys.last_value = value
         sys.last_traceback = tb
@@ -135,9 +139,8 @@ def showtraceback(self):
         """
         sys.last_type, sys.last_value, last_tb = ei = sys.exc_info()
         sys.last_traceback = last_tb
-        try:
-            # We remove the first stack item because it is our own code.
-            lines = traceback.format_exception(ei[0], ei[1], last_tb.tb_next)
-            self.write("".join(lines))
-        finally:
-            last_tb = ei = None
+        assert last_tb is not None
+
+        # We remove the first stack item because it is our own code.
+        lines = traceback.format_exception(ei[0], ei[1], last_tb.tb_next)
+        self.write("".join(lines))

synapse/util/ratelimitutils.py

@@ -15,33 +15,38 @@
 import collections
 import contextlib
 import logging
+from typing import DefaultDict
 
 from twisted.internet import defer
 
 from synapse.api.errors import LimitExceededError
+from synapse.config.ratelimiting import FederationRateLimitConfig
 from synapse.logging.context import (
     PreserveLoggingContext,
     make_deferred_yieldable,
     run_in_background,
 )
+from synapse.util import Clock
 
 logger = logging.getLogger(__name__)
 
 
 class FederationRateLimiter:
-    def __init__(self, clock, config):
+    def __init__(self, clock: Clock, config: FederationRateLimitConfig):
         """
         Args:
-            clock (Clock)
-            config (FederationRateLimitConfig)
+            clock
+            config
         """
 
         def new_limiter():
             return _PerHostRatelimiter(clock=clock, config=config)
 
-        self.ratelimiters = collections.defaultdict(new_limiter)
+        self.ratelimiters: DefaultDict[
+            str, "_PerHostRatelimiter"
+        ] = collections.defaultdict(new_limiter)
 
-    def ratelimit(self, host):
+    def ratelimit(self, host: str):
         """Used to ratelimit an incoming request from a given host
 
         Example usage:
@@ -79,7 +84,9 @@ def __init__(self, clock, config):
 
         # map from request_id object to Deferred for requests which are ready
         # for processing but have been queued
-        self.ready_request_queue = collections.OrderedDict()
+        self.ready_request_queue: collections.OrderedDict[
+            object, defer.Deferred
+        ] = collections.OrderedDict()
 
         # request id objects for requests which are in progress
         self.current_processing = set()
@@ -122,7 +129,7 @@ def _on_enter(self, request_id):
 
         def queue_request():
             if len(self.current_processing) >= self.concurrent_requests:
-                queue_defer = defer.Deferred()
+                queue_defer: defer.Deferred = defer.Deferred()
                 self.ready_request_queue[request_id] = queue_defer
                 logger.info(
                     "Ratelimiter: queueing request (queue now %i items)",

synapse/util/templates.py

@@ -27,7 +27,7 @@
 def build_jinja_env(
     template_search_directories: Iterable[str],
     config: "HomeServerConfig",
-    autoescape: Union[bool, Callable[[str], bool], None] = None,
+    autoescape: Union[bool, Callable[[Optional[str]], bool], None] = None,
 ) -> jinja2.Environment:
     """Set up a Jinja2 environment to load templates from the given search path
 
@@ -56,7 +56,8 @@ def build_jinja_env(
     if autoescape is None:
         autoescape = jinja2.select_autoescape()
 
-    loader = jinja2.FileSystemLoader(template_search_directories)
+    # the type signature of this is wrong
+    loader = jinja2.FileSystemLoader(template_search_directories)  # type: ignore[arg-type]
     env = jinja2.Environment(loader=loader, autoescape=autoescape)
 
     # Update the environment with our custom filters