tileradotorg/synapse
1
0
Fork 0
forked from MirrorHub/synapse
Code Pull requests Activity

Add type hints for ObservableDeferred attributes (#12159)

Browse source 
Signed-off-by: Sean Quah <seanq@element.io>
This commit is contained in:
Sean Quah 2022-03-04 15:37:02 +00:00 committed by GitHub
parent 158e0937eb
commit 75574726a7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/12159.misc Normal file
View file

@ -0,0 +1 @@
Add type hints for `ObservableDeferred` attributes.

14
synapse/util/async_helpers.py
View file

@ -40,7 +40,7 @@ from typing import (
)
import attr
from typing_extensions import ContextManager
from typing_extensions import ContextManager, Literal
from twisted.internet import defer
from twisted.internet.defer import CancelledError
@ -96,6 +96,10 @@ class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
__slots__ = ["_deferred", "_observers", "_result"]
_deferred: "defer.Deferred[_T]"
_observers: Union[List["defer.Deferred[_T]"], Tuple[()]]
_result: Union[None, Tuple[Literal[True], _T], Tuple[Literal[False], Failure]]
def __init__(self, deferred: "defer.Deferred[_T]", consumeErrors: bool = False):
object.__setattr__(self, "_deferred", deferred)
object.__setattr__(self, "_result", None)
@ -158,12 +162,14 @@ class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
effect the underlying deferred.
"""
if not self._result:
assert isinstance(self._observers, list)
d: "defer.Deferred[_T]" = defer.Deferred()
self._observers.append(d)
return d
elif self._result[0]:
return defer.succeed(self._result[1])
else:
success, res = self._result
return defer.succeed(res) if success else defer.fail(res)
return defer.fail(self._result[1])
def observers(self) -> "Collection[defer.Deferred[_T]]":
return self._observers
@ -175,6 +181,8 @@ class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
return self._result is not None and self._result[0] is True
def get_result(self) -> Union[_T, Failure]:
if self._result is None:
raise ValueError(f"{self!r} has no result yet")
return self._result[1]
def __getattr__(self, name: str) -> Any: