Fixes from typechecker

.github/workflows/build.yml

@@ -23,7 +23,7 @@ jobs:
       - name: Lint and coding style check with ruff
         run: ruff check --output-format=github .
       - name: Static type checking
-        run: mypy broqer --no-strict-optional --disable-error-code type-var --disable-error-code call-arg
+        run: mypy broqer
       - name: Check Readme style
         run: rstcheck README.rst
       - name: Test with pytest

CHANGELOG.md

@@ -1,3 +1,8 @@
+## 3.3.0
+
+* fixed bug in `Timer.end_early`
+* fixed typing annotations and extended `mypy` checks
+
 ## 3.2.0
 
 * added max queue threshold for CoroQueue (e.g., for `SinkAsync` and `MapAsync`)

CONTRIBUTING.rst

@@ -76,14 +76,13 @@ Ready to contribute? Here's how to set up `broqer` for local development.
 
    Now you can make your changes locally.
 
-5. When you're done making changes, check that your changes pass flake8 and the
-   tests, including testing other Python versions with tox::
+5. When you're done making changes, check that your changes pass ruff and the
+   tests, including static typechecking with mypy::
 
-    $ flake8 broqer tests
-    $ python setup.py test or py.test
-    $ tox
+    $ ruff check .
+    $ pytest
+    $ mypy
 
-   To get flake8 and tox, just pip install them into your virtualenv.
 
 6. Commit your changes and push your branch to GitHub::
 
@@ -102,8 +101,8 @@ Before you submit a pull request, check that it meets these guidelines:
 2. If the pull request adds functionality, the docs should be updated. Put
    your new functionality into a function with a docstring, and add the
    feature to the list in README.rst.
-3. The pull request should work for Python 3.6, 3.7, 3.8 and 3.9. Check
-   https://travis-ci.org/semiversus/python-broqer/pull_requests
+3. The pull request should work for Python 3.10, 3.11, 3.12, 3.13 and 3.14. Check
+   https://github.com/semiversus/python-broqer/pulls
    and make sure that the tests pass for all supported Python versions.
 
 Tips
@@ -118,11 +117,11 @@ Deploying
 ---------
 
 A reminder for the maintainers on how to deploy.
-Make sure all your changes are committed (including an entry in HISTORY.rst).
+Make sure all your changes are committed (including an entry in CHANGELOG.md).
 Then run::
 
 $ bumpversion patch # possible: major / minor / patch
 $ git push
 $ git push --tags
 
-Travis will then deploy to PyPI if tests pass.
+The GitHub Actions workflow will then deploy to PyPI if tests pass.

README.rst

@@ -24,15 +24,15 @@ Synopsis
 - Under MIT license (2018 Günther Jena)
 - Source is hosted on GitHub.com_
 - Documentation is hosted on ReadTheDocs.com_
-- Tested on Python 3.7. 3.8, 3.9, 3.10 and 3.11
-- Unit tested with pytest_, coding style checked with Flake8_, static type checked with mypy_, static code checked with Pylint_, documented with Sphinx_
+- Tested on Python 3.10, 3.11, 3.12, 3.13 and 3.14
+- Unit tested with pytest_, coding style checked with ruff_, static type checked with mypy_, static code checked with Pylint_, documented with Sphinx_
 - Operators known from ReactiveX_ and other streaming frameworks (like Map_, CombineLatest_, ...)
 
   + Centralised object to keep track of publishers and subscribers
   + Starting point to build applications with a microservice architecture
 
 .. _pytest: https://docs.pytest.org/en/latest
-.. _Flake8: http://flake8.pycqa.org/en/latest/
+.. _ruff: https://astral.sh/ruff
 .. _mypy: http://mypy-lang.org/
 .. _Pylint: https://www.pylint.org/
 .. _Sphinx: http://www.sphinx-doc.org

broqer/coro_queue.py

@@ -3,7 +3,8 @@
 import asyncio
 from collections import deque
 from enum import Enum
-from typing import Any, Deque, Optional, Tuple  # noqa: F401
+from typing import (Any, Awaitable, Callable, Deque,  # noqa: F401
+                    Optional, Tuple)
 from functools import partial
 
 from broqer import NONE
@@ -45,7 +46,8 @@ class CoroQueue:  # pylint: disable=too-few-public-methods
     :param max_queue_threshold: queue len error threshold,
                                 used with AsyncMode.QUEUE
     """
-    def __init__(self, coro, mode=AsyncMode.CONCURRENT,
+    def __init__(self, coro: Callable[..., Awaitable[Any]],
+                 mode: AsyncMode = AsyncMode.CONCURRENT,
                  max_queue_threshold: int | None = None):
 
         if max_queue_threshold is not None and mode != AsyncMode.QUEUE:
@@ -121,6 +123,12 @@ def _start_task(self, args: Tuple, future: asyncio.Future):
 
         # create a task out of it and add ._task_done as callback
         self._task = asyncio.ensure_future(self._coro(*args))
+        try:
+            self._task.set_name(f'CoroQueue({self._coro!r})')
+        except AttributeError:
+            # catch AttributeError if self._coro is a Future not a Coroutine
+            pass
+
         self._task.add_done_callback(partial(self._handle_done, future))
 
     def _handle_done(self, result_future: asyncio.Future, task: asyncio.Task):

broqer/op/combine_latest.py

@@ -23,7 +23,7 @@
 
 """
 from functools import wraps
-from typing import Any, Dict, MutableSequence, Callable  # noqa: F401
+from typing import Any, Dict, MutableSequence, Callable, Optional  # noqa: F401
 
 from broqer import Publisher, Subscriber, NONE
 
@@ -42,7 +42,8 @@ class CombineLatest(MultiOperator):
         state. emit_partial should only be used if an emit_on publisher is
         defined.
     """
-    def __init__(self, *publishers: Publisher, map_: Callable[..., Any] = None,
+    def __init__(self, *publishers: Publisher,
+                 map_: Optional[Callable[..., Any]] = None,
                  emit_on=None, emit_partial: bool = False) -> None:
         MultiOperator.__init__(self, *publishers)
 
@@ -133,7 +134,8 @@ def emit(self, value: Any, who: Publisher) -> None:
         return Publisher.notify(self, state)
 
 
-def build_combine_latest(map_: Callable[..., Any] = None, *, emit_on=None,
+def build_combine_latest(map_: Optional[Callable[..., Any]] = None, *,
+                         emit_on=None,
                          emit_partial: bool = False) -> Callable:
     """ Decorator to wrap a function to return a CombineLatest operator.
 

broqer/op/filter_.py

@@ -26,7 +26,7 @@
 
 """
 from functools import partial, wraps
-from typing import Any, Callable
+from typing import Any, Callable, Optional
 
 from broqer import NONE, Publisher
 from broqer.operator import Operator
@@ -83,7 +83,7 @@ class EvalTrue(Operator):
     This operator can be used in the pipline style (v | EvalTrue()) or as
     standalone operation (EvalTrue(v)).
     """
-    def __init__(self, publisher: Publisher = None) -> None:
+    def __init__(self, publisher: Optional[Publisher] = None) -> None:
         Operator.__init__(self)
         self._originator = publisher
 
@@ -114,7 +114,7 @@ class EvalFalse(Operator):
 
     This operator can be used in the pipline style (v | EvalFalse() or as
     standalone operation (EvalFalse(v))."""
-    def __init__(self, publisher: Publisher = None) -> None:
+    def __init__(self, publisher: Optional[Publisher] = None) -> None:
         Operator.__init__(self)
         self._originator = publisher
 
@@ -140,7 +140,7 @@ def emit(self, value: Any, who: Publisher) -> None:
         return None
 
 
-def build_filter(predicate: Callable[[Any], bool] = None, *,
+def build_filter(predicate: Optional[Callable[[Any], bool]] = None, *,
                  unpack: bool = False):
     """ Decorator to wrap a function to return a Filter operator.
 
@@ -156,7 +156,7 @@ def _build_filter(predicate):
     return _build_filter
 
 
-def build_filter_factory(predicate: Callable[[Any], bool] = None, *,
+def build_filter_factory(predicate: Optional[Callable[[Any], bool]] = None, *,
                          unpack: bool = False):
     """ Decorator to wrap a function to return a factory for Filter operators.
 

broqer/op/map_.py

@@ -36,7 +36,7 @@
 EMITTED None
 """
 from functools import partial, wraps
-from typing import Any, Callable
+from typing import Any, Callable, Optional
 
 from broqer import Publisher, NONE
 from broqer.publisher import ValueT
@@ -51,7 +51,7 @@ class Map(Operator):
     :param unpack: value from emits will be unpacked (\\*value)
     :param \\*\\*kwargs: keyword arguments to be used for calling function
     """
-    def __init__(self, function: Callable[[Any], Any], *args,
+    def __init__(self, function: Callable[..., Any], *args,
                  unpack: bool = False, **kwargs) -> None:
         """ Special care for return values:
               - return `None` (or nothing) if you don't want to return a result
@@ -98,7 +98,7 @@ def emit(self, value: ValueT, who: Publisher) -> None:
         return None
 
 
-def build_map(function: Callable[..., None] = None, *,
+def build_map(function: Optional[Callable[..., Any]] = None, *,
               unpack: bool = False):
     """ Decorator to wrap a function to return a Map operator.
 
@@ -114,14 +114,14 @@ def _build_map(function):
     return _build_map
 
 
-def build_map_factory(function: Callable[[Any], Any] = None,
+def build_map_factory(function: Optional[Callable[..., Any]] = None,
                       unpack: bool = False):
     """ Decorator to wrap a function to return a factory for Map operators.
 
     :param function: function to be wrapped
     :param unpack: value from emits will be unpacked (*value)
     """
-    def _build_map(function: Callable[[Any], Any]):
+    def _build_map(function: Callable[..., Any]):
         @wraps(function)
         def _wrapper(*args, **kwargs) -> Map:
             if 'unpack' in kwargs:

broqer/publisher.py

@@ -1,7 +1,7 @@
 """ Implementing Publisher """
 import sys
-from typing import (TYPE_CHECKING, TypeVar, Type, Tuple, Callable, Optional,
-                    overload)
+from typing import (TYPE_CHECKING, Any, TypeVar, Type, Tuple, Callable,
+                    Optional, overload)
 
 from broqer import NONE, Disposable, default_error_handler
 import broqer
@@ -55,14 +55,15 @@ class Publisher:
                          indirectly) dependent on.
     """
     @overload  # noqa: F811
-    def __init__(self, *, type_: Type[ValueT] = None):
+    def __init__(self, *, type_: Optional[Type[ValueT]] = None):
         pass
 
     @overload  # noqa: F811
-    def __init__(self, init: ValueT, type_: Type[ValueT] = None):  # noqa: F811
+    def __init__(self, init: ValueT,  # noqa: F811
+                 type_: Optional[Type[ValueT]] = None):
         pass
 
-    def __init__(self, init=NONE, type_=None):  # noqa: F811
+    def __init__(self, init=NONE, type_: Optional[Type] = None):  # noqa: F811
         self._state = init
 
         if type_:
@@ -164,8 +165,9 @@ def subscriptions(self) -> Tuple['Subscriber', ...]:
         """ Property returning a tuple with all current subscribers """
         return tuple(self._subscriptions)
 
-    def register_on_subscription_callback(self,
-                                          callback: SubscriptionCBT) -> None:
+    def register_on_subscription_callback(
+        self, callback: Optional[SubscriptionCBT]
+    ) -> None:
         """ This callback will be called, when the subscriptions are changing.
         When a subscription is done and no subscription was present the
         callback is called with True as argument. When after unsubscribe no
@@ -194,8 +196,8 @@ def __await__(self):
         future = self.as_future(timeout=None, omit_subscription=False)
         return future.__await__()
 
-    def as_future(self, timeout: float, omit_subscription: bool = True,
-                  loop=None):
+    def as_future(self, timeout: Optional[float],
+                  omit_subscription: bool = True, loop=None):
         """ Returns a asyncio.Future which will be done on first change of this
         publisher.
 
@@ -250,6 +252,67 @@ def __or__(self, operator: 'Operator'):
         operator.originator = self
         return operator
 
+    if TYPE_CHECKING:
+        # These are declaration-only and carry no runtime cost. Keep this list
+        # in sync with `apply_operator_overloading()`.
+        #
+        # Deliberately NOT declared here:
+        #   __or__      - statically defined above (operator piping); it is
+        #                 not monkey-patched.
+        #   __getattr__ - installed at runtime, but only succeeds when
+        #                 `inherited_type` is set. Declaring it would type
+        #                 *every* attribute access on a Publisher as valid and
+        #                 silently swallow genuine typos.
+
+        # binary operators
+        def __lt__(self, other: Any) -> 'Publisher': ...
+        def __le__(self, other: Any) -> 'Publisher': ...
+        def __eq__(self, other: Any) -> 'Publisher': ...  # type: ignore[override]
+        def __ne__(self, other: Any) -> 'Publisher': ...  # type: ignore[override]
+        def __ge__(self, other: Any) -> 'Publisher': ...
+        def __gt__(self, other: Any) -> 'Publisher': ...
+        def __add__(self, other: Any) -> 'Publisher': ...
+        def __and__(self, other: Any) -> 'Publisher': ...
+        def __lshift__(self, other: Any) -> 'Publisher': ...
+        def __mod__(self, other: Any) -> 'Publisher': ...
+        def __mul__(self, other: Any) -> 'Publisher': ...
+        def __pow__(self, other: Any) -> 'Publisher': ...
+        def __rshift__(self, other: Any) -> 'Publisher': ...
+        def __sub__(self, other: Any) -> 'Publisher': ...
+        def __xor__(self, other: Any) -> 'Publisher': ...
+        def __concat__(self, other: Any) -> 'Publisher': ...
+        def __getitem__(self, key: Any) -> 'Publisher': ...
+        def __floordiv__(self, other: Any) -> 'Publisher': ...
+        def __truediv__(self, other: Any) -> 'Publisher': ...
+
+        # # reflected binary operators
+        def __radd__(self, other: Any) -> 'Publisher': ...
+        def __rand__(self, other: Any) -> 'Publisher': ...
+        def __rlshift__(self, other: Any) -> 'Publisher': ...
+        def __rmod__(self, other: Any) -> 'Publisher': ...
+        def __rmul__(self, other: Any) -> 'Publisher': ...
+        def __rpow__(self, other: Any) -> 'Publisher': ...
+        def __rrshift__(self, other: Any) -> 'Publisher': ...
+        def __rsub__(self, other: Any) -> 'Publisher': ...
+        def __rxor__(self, other: Any) -> 'Publisher': ...
+        def __rfloordiv__(self, other: Any) -> 'Publisher': ...
+        def __rtruediv__(self, other: Any) -> 'Publisher': ...
+
+        # unary operators
+        def __neg__(self) -> 'Publisher': ...
+        def __pos__(self) -> 'Publisher': ...
+        def __abs__(self) -> 'Publisher': ...
+        def __invert__(self) -> 'Publisher': ...
+        def __round__(self, ndigits: Any = None) -> 'Publisher': ...  # type: ignore[override]
+        def __trunc__(self) -> 'Publisher': ...  # type: ignore[override]
+        def __floor__(self) -> 'Publisher': ...
+        def __ceil__(self) -> 'Publisher': ...
+
+        # Declaring __eq__ in a class body would otherwise set __hash__ to
+        # None. Publisher keeps identity hashing (it is used as a dict key
+        # in CombineLatest._index).
+        __hash__ = object.__hash__
+
     def __dir__(self):
         """ Extending __dir__ with inherited type """
         attrs = set(super().__dir__())

broqer/publishers/poll.py

@@ -15,7 +15,7 @@ class PollPublisher(Publisher):
     :param interval: Time in seconds between polling calls
     """
     def __init__(self, poll_cb: Callable[[], Any], interval: float, *,
-                 type_: Type[ValueT] = None):
+                 type_: Optional[Type[ValueT]] = None):
         Publisher.__init__(self, type_=type_)
         self.poll_cb = poll_cb
         self.interval = interval