Add some simple type hints on Deferreds

synapse/http/client.py

@@ -847,7 +847,7 @@ def connectionLost(self, reason: Failure = connectionDone) -> None:
 
 def read_body_with_max_size(
     response: IResponse, stream: ByteWriteable, max_size: Optional[int]
-) -> defer.Deferred:
+) -> defer.Deferred[int]:
     """
     Read a HTTP response body to a file-object. Optionally enforcing a maximum file size.
 
@@ -862,7 +862,7 @@ def read_body_with_max_size(
     Returns:
         A Deferred which resolves to the length of the read body.
     """
-    d = defer.Deferred()
+    d: defer.Deferred[int] = defer.Deferred()
 
     # If the Content-Length header gives a size larger than the maximum allowed
     # size, do not bother downloading the body.

synapse/replication/tcp/client.py

@@ -285,7 +285,7 @@ async def wait_for_stream_position(
 
         # Create a new deferred that times out after N seconds, as we don't want
         # to wedge here forever.
-        deferred = Deferred()
+        deferred: Deferred[None] = Deferred()
         deferred = timeout_deferred(
             deferred, _WAIT_FOR_REPLICATION_TIMEOUT_SECONDS, self._reactor
         )

synapse/util/async_helpers.py

@@ -49,6 +49,8 @@
 
 logger = logging.getLogger(__name__)
 
+_T = TypeVar("_T")
+
 
 class ObservableDeferred:
     """Wraps a deferred object so that we can add observer deferreds. These
@@ -121,7 +123,7 @@ def observe(self) -> defer.Deferred:
         effect the underlying deferred.
         """
         if not self._result:
-            d = defer.Deferred()
+            d: defer.Deferred[Any] = defer.Deferred()
 
             def remove(r):
                 self._observers.discard(d)
@@ -415,7 +417,7 @@ def __init__(self):
         self.key_to_current_writer: Dict[str, defer.Deferred] = {}
 
     async def read(self, key: str) -> ContextManager:
-        new_defer = defer.Deferred()
+        new_defer: defer.Deferred[None] = defer.Deferred()
 
         curr_readers = self.key_to_current_readers.setdefault(key, set())
         curr_writer = self.key_to_current_writer.get(key, None)
@@ -438,7 +440,7 @@ def _ctx_manager():
         return _ctx_manager()
 
     async def write(self, key: str) -> ContextManager:
-        new_defer = defer.Deferred()
+        new_defer: defer.Deferred[None] = defer.Deferred()
 
         curr_readers = self.key_to_current_readers.get(key, set())
         curr_writer = self.key_to_current_writer.get(key, None)
@@ -471,10 +473,8 @@ def _ctx_manager():
 
 
 def timeout_deferred(
-    deferred: defer.Deferred,
-    timeout: float,
-    reactor: IReactorTime,
-) -> defer.Deferred:
+    deferred: defer.Deferred[_T], timeout: float, reactor: IReactorTime
+) -> defer.Deferred[_T]:
     """The in built twisted `Deferred.addTimeout` fails to time out deferreds
     that have a canceller that throws exceptions. This method creates a new
     deferred that wraps and times out the given deferred, correctly handling
@@ -497,7 +497,7 @@ def timeout_deferred(
     Returns:
         A new Deferred, which will errback with defer.TimeoutError on timeout.
     """
-    new_d = defer.Deferred()
+    new_d: defer.Deferred[_T] = defer.Deferred()
 
     timed_out = [False]
 

synapse/util/caches/descriptors.py

@@ -413,7 +413,7 @@ def arg_to_cache_key(arg):
                 # relevant result for that key.
                 deferreds_map = {}
                 for arg in missing:
-                    deferred = defer.Deferred()
+                    deferred: defer.Deferred[Any] = defer.Deferred()
                     deferreds_map[arg] = deferred
                     key = arg_to_cache_key(arg)
                     cache.set(key, deferred, callback=invalidate_callback)