agronholm / apscheduler

Task scheduling library for Python
MIT License
5.98k stars 694 forks source link

Typing of scheduler subscribe() method #846

Closed peterschutt closed 2 months ago

peterschutt commented 6 months ago

Things to check first

Feature description

Propose using a bound typevar for AsyncScheduler.subscribe():

diff --git a/src/apscheduler/_schedulers/async_.py b/src/apscheduler/_schedulers/async_.py
index 5363dd3..963e09d 100644
--- a/src/apscheduler/_schedulers/async_.py
+++ b/src/apscheduler/_schedulers/async_.py
@@ -11,7 +11,7 @@ from functools import partial
 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, TypeVar, cast
 from uuid import UUID, uuid4

 import anyio
@@ -72,6 +72,7 @@ _microsecond_delta = timedelta(microseconds=1)
 _zero_timedelta = timedelta()

 TaskType: TypeAlias = "Task | str | Callable"
+EventType = TypeVar("EventType", bound="Event")

 @attrs.define(eq=False)
@@ -216,8 +217,8 @@ class AsyncScheduler:

     def subscribe(
         self,
-        callback: Callable[[Event], Any],
-        event_types: type[Event] | Iterable[type[Event]] | None = None,
+        callback: Callable[[EventType], Any],
+        event_types: type[EventType] | Iterable[type[EventType]] | None = None,
         *,
         one_shot: bool = False,
         is_async: bool = True,

Use case

Curerently, a subscription like this:

scheduler = AsyncScheduler()

def handler(event: JobAcquired) -> None:
    pass

scheduler.subscribe(handler, JobAcquired)

Raises a type error:

error: Argument 1 to "subscribe" of "AsyncScheduler" has incompatible type "Callable[[JobAcquired], None]"; expected "Callable[[Event], Any]"  [arg-type]

With proposed changeset, there is no error.

This may have additional benefit where a handler that cannot handle the subscribed types will be picked up statically, e.g:

scheduler = AsyncScheduler()

def handler(event: JobReleased) -> None:
    pass

scheduler.subscribe(handler, JobAcquired)

With above patch applied, mypy raises:

error: Argument 1 to "subscribe" of "AsyncScheduler" has incompatible type "Callable[[JobReleased], None]"; expected "Callable[[JobAcquired], Any]"  [arg-type]
agronholm commented 6 months ago

Could you make this a PR instead?

peterschutt commented 6 months ago

Could you make this a PR instead?

yep, will do.