Enhanced typing of subscribe() (#847)

This PR adds type var `T_Event` and uses it to type the `subscribe()` interfaces.

Fixes an issue where subscribing a handler would raise a type error if the handler is typed to receive a subclass of `Event`.

Allows type checkers to verify that the handler passed to `subscribe()` can support the event types it is assigned to handle.

Closes #846.

pyproject.toml

@@ -54,12 +54,12 @@ test = [
     "anyio[trio]",
     "asyncmy >= 0.2.5; python_implementation == 'CPython'",
     "coverage >= 7",
-    "paho-mqtt >= 2.0",
+    "paho-mqtt >= 1.5",
     "psycopg",
     "pymongo >= 4",
     "pymysql[rsa]",
     "PySide6 >= 6.6; python_implementation == 'CPython'",
-    "pytest >= 8.0",
+    "pytest >= 7.4",
     "pytest-lazy-fixtures",
     "pytest-mock",
     "time-machine >= 2.13.0; python_implementation == 'CPython'",
@@ -132,4 +132,7 @@ commands = pyright --verifytypes apscheduler
 [testenv:docs]
 extras = doc
 commands = sphinx-build -n docs build/sphinx
+
+[testenv:paho-mqtt-v1]
+deps = paho-mqtt~=1.5
 """

src/apscheduler/_events.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timezone
 from functools import partial
 from traceback import format_tb
-from typing import Any
+from typing import Any, TypeVar
 from uuid import UUID
 
 import attrs
@@ -14,6 +14,8 @@
 from ._structures import Job, JobResult
 from ._utils import qualified_name
 
+T_Event = TypeVar("T_Event", bound="Event")
+
 
 @attrs.define(kw_only=True, frozen=True)
 class Event:

src/apscheduler/_schedulers/async_.py

@@ -11,7 +11,7 @@
 from inspect import isbuiltin, isclass, ismethod, ismodule
 from logging import Logger, getLogger
 from types import TracebackType
-from typing import Any, Callable, Iterable, Mapping, cast
+from typing import Any, Callable, Iterable, Mapping, cast, overload
 from uuid import UUID, uuid4
 
 import anyio
@@ -38,6 +38,7 @@
     SchedulerStarted,
     SchedulerStopped,
     ScheduleUpdated,
+    T_Event,
 )
 from .._exceptions import (
     CallableLookupError,
@@ -214,10 +215,30 @@ async def cleanup(self) -> None:
         await self.data_store.cleanup()
         self.logger.info("Cleaned up expired job results and finished schedules")
 
+    @overload
+    def subscribe(
+        self,
+        callback: Callable[[T_Event], Any],
+        event_types: type[T_Event],
+        *,
+        one_shot: bool = ...,
+        is_async: bool = ...,
+    ) -> Subscription: ...
+
+    @overload
     def subscribe(
         self,
         callback: Callable[[Event], Any],
-        event_types: type[Event] | Iterable[type[Event]] | None = None,
+        event_types: Iterable[type[Event]] | None = None,
+        *,
+        one_shot: bool = False,
+        is_async: bool = True,
+    ) -> Subscription: ...
+
+    def subscribe(
+        self,
+        callback: Callable[[T_Event], Any],
+        event_types: type[T_Event] | Iterable[type[T_Event]] | None = None,
         *,
         one_shot: bool = False,
         is_async: bool = True,

src/apscheduler/_schedulers/sync.py

@@ -10,13 +10,14 @@
 from functools import partial
 from logging import Logger
 from types import TracebackType
-from typing import Any, Callable, Iterable, Mapping
+from typing import Any, Callable, Iterable, Mapping, overload
 from uuid import UUID
 
 from anyio.from_thread import BlockingPortal, start_blocking_portal
 
-from .. import Event, current_scheduler
+from .. import current_scheduler
 from .._enums import CoalescePolicy, ConflictPolicy, RunState, SchedulerRole
+from .._events import Event, T_Event
 from .._structures import Job, JobResult, Schedule, Task
 from .._utils import UnsetValue, unset
 from ..abc import DataStore, EventBroker, JobExecutor, Subscription, Trigger
@@ -155,12 +156,30 @@ def cleanup(self) -> None:
         self._ensure_services_ready()
         return self._portal.call(self._async_scheduler.cleanup)
 
+    @overload
+    def subscribe(
+        self,
+        callback: Callable[[T_Event], Any],
+        event_types: type[T_Event],
+        *,
+        one_shot: bool = ...,
+    ) -> Subscription: ...
+
+    @overload
     def subscribe(
         self,
         callback: Callable[[Event], Any],
         event_types: Iterable[type[Event]] | None = None,
         *,
         one_shot: bool = False,
+    ) -> Subscription: ...
+
+    def subscribe(
+        self,
+        callback: Callable[[T_Event], Any],
+        event_types: type[T_Event] | Iterable[type[T_Event]] | None = None,
+        *,
+        one_shot: bool = False,
     ) -> Subscription:
         """
         Subscribe to events.

src/apscheduler/abc.py

@@ -15,7 +15,7 @@
 
 if TYPE_CHECKING:
     from ._enums import ConflictPolicy
-    from ._events import Event
+    from ._events import Event, T_Event
     from ._structures import Job, JobResult, Schedule, Task
 
 
@@ -135,8 +135,8 @@ async def publish_local(self, event: Event) -> None:
     @abstractmethod
     def subscribe(
         self,
-        callback: Callable[[Event], Any],
-        event_types: Iterable[type[Event]] | None = None,
+        callback: Callable[[T_Event], Any],
+        event_types: Iterable[type[T_Event]] | None = None,
         *,
         is_async: bool = True,
         one_shot: bool = False,