Disallow untyped defs (#632)

.pre-commit-config.yaml

@@ -26,7 +26,7 @@ repos:
         language: node
         pass_filenames: false
         types: [python]
-        additional_dependencies: ["pyright@1.1.228"]
+        additional_dependencies: ["pyright@1.1.229"]
     repo: local
   - hooks:
       - id: mypy

examples/parallel/timer.py

@@ -7,12 +7,12 @@
 seconds = [5, 1, 2, 4, 3]
 
 
-def sleep(tm):
+def sleep(tm: float) -> float:
     time.sleep(tm)
     return tm
 
 
-def output(result):
+def output(result: str) -> None:
     print("%d seconds" % result)
 
 

pyproject.toml

@@ -64,6 +64,8 @@ python_version = "3.9"
 follow_imports = "silent"
 files = ["reactivex"]
 exclude = ["reactivex/operators/_\\w.*\\.py$"] # mypy will eventually catch up
+disallow_any_generics = true
+disallow_untyped_defs = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

reactivex/abc/disposable.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from types import TracebackType
 from typing import Optional, Type
@@ -15,15 +17,16 @@ def dispose(self) -> None:
         """
         raise NotImplementedError
 
-    def __enter__(self):
+    def __enter__(self) -> DisposableBase:
         """Context management protocol."""
+        return self
 
     def __exit__(
         self,
         exctype: Optional[Type[BaseException]],
         excinst: Optional[BaseException],
         exctb: Optional[TracebackType],
-    ):
+    ) -> None:
         """Context management protocol."""
         self.dispose()
 

reactivex/disposable/booleandisposable.py

@@ -6,7 +6,7 @@
 class BooleanDisposable(DisposableBase):
     """Represents a Disposable that can be checked for status."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initializes a new instance of the BooleanDisposable class."""
 
         self.is_disposed = False

reactivex/disposable/multipleassignmentdisposable.py

@@ -8,7 +8,7 @@ class MultipleAssignmentDisposable(DisposableBase):
     """Represents a disposable resource whose underlying disposable
     resource can be replaced by another disposable resource."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.current: Optional[DisposableBase] = None
         self.is_disposed = False
         self.lock = RLock()

reactivex/disposable/scheduleddisposable.py

@@ -30,7 +30,7 @@ def is_disposed(self) -> bool:
     def dispose(self) -> None:
         """Disposes the wrapped disposable on the provided scheduler."""
 
-        def action(scheduler: abc.SchedulerBase, state: Any):
+        def action(scheduler: abc.SchedulerBase, state: Any) -> None:
             """Scheduled dispose action"""
 
             self.disposable.dispose()

reactivex/disposable/singleassignmentdisposable.py

@@ -25,7 +25,7 @@ def __init__(self) -> None:
     def get_disposable(self) -> Optional[DisposableBase]:
         return self.current
 
-    def set_disposable(self, value: DisposableBase):
+    def set_disposable(self, value: DisposableBase) -> None:
         if self.current:
             raise Exception("Disposable has already been assigned")
 

reactivex/internal/basic.py

@@ -4,7 +4,7 @@
 _T = TypeVar("_T")
 
 
-def noop(*args: Any, **kw: Any):
+def noop(*args: Any, **kw: Any) -> None:
     """No operation. Returns nothing"""
 
 

reactivex/internal/utils.py

@@ -20,7 +20,7 @@ def add_ref(xs: "Observable[_T]", r: RefCountDisposable) -> "Observable[_T]":
 
     def subscribe(
         observer: abc.ObserverBase[Any], scheduler: Optional[abc.SchedulerBase] = None
-    ):
+    ) -> abc.DisposableBase:
         return CompositeDisposable(r.disposable, xs.subscribe(observer))
 
     return Observable(subscribe)

reactivex/notification.py

@@ -81,7 +81,7 @@ def subscribe(
             observer: abc.ObserverBase[_T],
             scheduler: Optional[abc.SchedulerBase] = None,
         ) -> abc.DisposableBase:
-            def action(scheduler: abc.SchedulerBase, state: Any):
+            def action(scheduler: abc.SchedulerBase, state: Any) -> None:
                 self._accept_observer(observer)
                 if self.kind == "N":
                     observer.on_completed()
@@ -105,7 +105,7 @@ def __eq__(self, other: Any) -> bool:
 class OnNext(Notification[_T]):
     """Represents an OnNext notification to an observer."""
 
-    def __init__(self, value: _T):
+    def __init__(self, value: _T) -> None:
         """Constructs a notification of a new value."""
 
         super(OnNext, self).__init__()
@@ -134,7 +134,7 @@ def __str__(self) -> str:
 class OnError(Notification[_T]):
     """Represents an OnError notification to an observer."""
 
-    def __init__(self, error: Union[Exception, str]):
+    def __init__(self, error: Union[Exception, str]) -> None:
         """Constructs a notification of an exception."""
 
         super(OnError, self).__init__()
@@ -151,7 +151,7 @@ def _accept(
     ) -> None:
         return on_error(self.exception) if on_error else None
 
-    def _accept_observer(self, observer: abc.ObserverBase[_T]):
+    def _accept_observer(self, observer: abc.ObserverBase[_T]) -> None:
         return observer.on_error(self.exception)
 
     def __str__(self) -> str:
@@ -161,7 +161,7 @@ def __str__(self) -> str:
 class OnCompleted(Notification[_T]):
     """Represents an OnCompleted notification to an observer."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Constructs a notification of the end of a sequence."""
 
         super(OnCompleted, self).__init__()
@@ -175,7 +175,7 @@ def _accept(
     ) -> None:
         return on_completed() if on_completed else None
 
-    def _accept_observer(self, observer: abc.ObserverBase[_T]):
+    def _accept_observer(self, observer: abc.ObserverBase[_T]) -> None:
         return observer.on_completed()
 
     def __str__(self) -> str:
@@ -196,10 +196,10 @@ def from_notifier(handler: Callable[[Notification[_T]], None]) -> Observer[_T]:
     def _on_next(value: _T) -> None:
         return handler(OnNext(value))
 
-    def _on_error(error: Exception):
+    def _on_error(error: Exception) -> None:
         return handler(OnError(error))
 
-    def _on_completed():
+    def _on_completed() -> None:
         return handler(OnCompleted())
 
     return Observer(_on_next, _on_error, _on_completed)

reactivex/observable/amb.py

@@ -19,7 +19,7 @@ def amb_(*sources: Observable[_T]) -> Observable[_T]:
 
     acc: Observable[_T] = never()
 
-    def func(previous: Observable[_T], current: Observable[_T]):
+    def func(previous: Observable[_T], current: Observable[_T]) -> Observable[_T]:
         return _.amb(previous)(current)
 
     for source in sources:

reactivex/observable/catch.py

@@ -34,15 +34,15 @@ def catch_with_iterable_(sources: Iterable[Observable[_T]]) -> Observable[_T]:
 
     def subscribe(
         observer: abc.ObserverBase[_T], scheduler_: Optional[abc.SchedulerBase] = None
-    ):
+    ) -> abc.DisposableBase:
         _scheduler = scheduler_ or CurrentThreadScheduler.singleton()
 
         subscription = SerialDisposable()
         cancelable = SerialDisposable()
         last_exception = None
         is_disposed = False
 
-        def action(scheduler: abc.SchedulerBase, state: Any = None):
+        def action(scheduler: abc.SchedulerBase, state: Any = None) -> None:
             def on_error(exn: Exception) -> None:
                 nonlocal last_exception
                 last_exception = exn
@@ -72,7 +72,7 @@ def on_error(exn: Exception) -> None:
 
         cancelable.disposable = _scheduler.schedule(action)
 
-        def dispose():
+        def dispose() -> None:
             nonlocal is_disposed
             is_disposed = True
 

reactivex/observable/combinelatest.py

@@ -29,7 +29,7 @@ def subscribe(
         is_done = [False] * n
         values = [None] * n
 
-        def _next(i: Any):
+        def _next(i: Any) -> None:
             has_value[i] = True
 
             if has_value_all[0] or all(has_value):
@@ -41,22 +41,22 @@ def _next(i: Any):
 
             has_value_all[0] = all(has_value)
 
-        def done(i: Any):
+        def done(i: Any) -> None:
             is_done[i] = True
             if all(is_done):
                 observer.on_completed()
 
         subscriptions: List[Optional[SingleAssignmentDisposable]] = [None] * n
 
-        def func(i: int):
+        def func(i: int) -> None:
             subscriptions[i] = SingleAssignmentDisposable()
 
             def on_next(x: Any) -> None:
                 with parent.lock:
                     values[i] = x
                     _next(i)
 
-            def on_completed():
+            def on_completed() -> None:
                 with parent.lock:
                     done(i)
 

reactivex/observable/concat.py

@@ -24,12 +24,12 @@ def subscribe(
         cancelable = SerialDisposable()
         is_disposed = False
 
-        def action(scheduler: abc.SchedulerBase, state: Any = None):
+        def action(scheduler: abc.SchedulerBase, state: Any = None) -> None:
             nonlocal is_disposed
             if is_disposed:
                 return
 
-            def on_completed():
+            def on_completed() -> None:
                 cancelable.disposable = _scheduler.schedule(action)
 
             try:
@@ -50,7 +50,7 @@ def on_completed():
 
         cancelable.disposable = _scheduler.schedule(action)
 
-        def dispose():
+        def dispose() -> None:
             nonlocal is_disposed
             is_disposed = True
 

reactivex/observable/connectableobservable.py

@@ -24,10 +24,12 @@ def _subscribe_core(
         self,
         observer: abc.ObserverBase[_T],
         scheduler: Optional[abc.SchedulerBase] = None,
-    ):
+    ) -> abc.DisposableBase:
         return self.subject.subscribe(observer, scheduler=scheduler)
 
-    def connect(self, scheduler: Optional[abc.SchedulerBase] = None):
+    def connect(
+        self, scheduler: Optional[abc.SchedulerBase] = None
+    ) -> Optional[abc.DisposableBase]:
         """Connects the observable."""
 
         if not self.has_subscription:
@@ -70,7 +72,7 @@ def subscribe(
                 connectable_subscription[0] = source.connect(scheduler)
                 is_connected[0] = True
 
-            def dispose():
+            def dispose() -> None:
                 subscription.dispose()
                 count[0] -= 1
                 is_connected[0] = False

reactivex/observable/forkjoin.py

@@ -46,7 +46,7 @@ def done(i: int) -> None:
             cast(SingleAssignmentDisposable, None)
         ] * n
 
-        def _subscribe(i: int):
+        def _subscribe(i: int) -> None:
             subscriptions[i] = SingleAssignmentDisposable()
 
             def on_next(value: Any) -> None:

reactivex/observable/fromcallback.py

@@ -22,14 +22,14 @@ def from_callback_(
         the arguments to the callback as a list.
     """
 
-    def function(*args: Any):
+    def function(*args: Any) -> Observable[Any]:
         arguments = list(args)
 
         def subscribe(
             observer: abc.ObserverBase[Any],
             scheduler: Optional[abc.SchedulerBase] = None,
         ) -> abc.DisposableBase:
-            def handler(*args: Any):
+            def handler(*args: Any) -> None:
                 results = list(args)
                 if mapper:
                     try:

reactivex/observable/fromfuture.py

@@ -23,7 +23,7 @@ def from_future_(future: "Future[_T]") -> Observable[_T]:
     def subscribe(
         observer: abc.ObserverBase[Any], scheduler: Optional[abc.SchedulerBase] = None
     ) -> abc.DisposableBase:
-        def done(future: "Future[_T]"):
+        def done(future: "Future[_T]") -> None:
             try:
                 value: Any = future.result()
             except Exception as ex:

reactivex/observable/generate.py

@@ -21,7 +21,7 @@ def subscribe(
         state = initial_state
         mad = MultipleAssignmentDisposable()
 
-        def action(scheduler: abc.SchedulerBase, state1: Any = None):
+        def action(scheduler: abc.SchedulerBase, state1: Any = None) -> None:
             nonlocal first
             nonlocal state
 

reactivex/observable/generatewithrelativetime.py

@@ -39,7 +39,7 @@ def generate_with_relative_time_(
     def subscribe(
         observer: abc.ObserverBase[_TState],
         scheduler: Optional[abc.SchedulerBase] = None,
-    ):
+    ) -> abc.DisposableBase:
         scheduler = scheduler or TimeoutScheduler.singleton()
         mad = MultipleAssignmentDisposable()
         state = initial_state
@@ -48,7 +48,7 @@ def subscribe(
         first = True
         time: Optional[RelativeTime] = None
 
-        def action(scheduler: abc.SchedulerBase, _: Any):
+        def action(scheduler: abc.SchedulerBase, _: Any) -> None:
             nonlocal state
             nonlocal has_result
             nonlocal result