use typevar to control callback argument variance

Author: eternal-sorrow

i3ipc/_private/__init__.py

@@ -1,3 +1,3 @@
-from .pubsub import PubSub
+from .pubsub import PubSub, Handler
 from .types import MessageType, ReplyType, EventType
 from .sync import Synchronizer

i3ipc/_private/pubsub.py

@@ -1,11 +1,13 @@
-from typing import Callable, Optional, TypeAlias, TypedDict, TYPE_CHECKING
+from typing import Callable, Optional, TypeAlias, TypedDict, TypeVar, TYPE_CHECKING
 
 from i3ipc.events import IpcBaseEvent
 
 if TYPE_CHECKING:
     from i3ipc.connection import Connection
 
-Handler: TypeAlias = Callable[['Connection', IpcBaseEvent], None]
+_BaseEvent = TypeVar('_BaseEvent', bound=IpcBaseEvent, contravariant=True)
+
+Handler: TypeAlias = Callable[['Connection', _BaseEvent], None]
 
 
 class Subscription(TypedDict):

i3ipc/aio/connection.py

@@ -6,7 +6,7 @@
 from .. import con
 import os
 import json
-from typing import Callable, Coroutine, Optional, TypeAlias, TypedDict, Union
+from typing import Callable, Coroutine, Optional, TypeAlias, TypedDict, TypeVar, Union
 import struct
 import socket
 import logging
@@ -33,8 +33,11 @@ def ensure_future(obj):
     return future
 
 
+_BaseEvent = TypeVar('_BaseEvent', bound=IpcBaseEvent, contravariant=True)
+
+
 Handler: TypeAlias = Union[
-    Callable[['Connection', IpcBaseEvent], None], Callable[['Connection', IpcBaseEvent], Coroutine]
+    Callable[['Connection', _BaseEvent], None], Callable[['Connection', _BaseEvent], Coroutine]
 ]
 
 
@@ -562,7 +565,7 @@ def _on(self, event: Union[Event, str], handler: Handler):
         self._pubsub.subscribe(event, handler)  # type: ignore[arg-type]
         ensure_future(self.subscribe([base_event]))
 
-    def off(self, handler: Callable[['Connection', IpcBaseEvent], None]):
+    def off(self, handler: Handler):
         """Unsubscribe the handler from being called on ipc events.
 
         :param handler: The handler that was previously attached with

i3ipc/connection.py

@@ -5,9 +5,9 @@
                       VersionReply, WorkspaceReply, SeatReply, InputReply)
 from .events import (IpcBaseEvent, BarconfigUpdateEvent, BindingEvent, OutputEvent, ShutdownEvent,
                      WindowEvent, TickEvent, ModeEvent, WorkspaceEvent, InputEvent, Event)
-from ._private import PubSub, MessageType, EventType, Synchronizer
+from ._private import PubSub, Handler, MessageType, EventType, Synchronizer
 
-from typing import List, Optional, Union, Callable
+from typing import List, Optional, Union
 import struct
 import json
 import socket
@@ -380,7 +380,7 @@ def _subscribe(self, events):
         self.subscriptions |= events
         return result
 
-    def off(self, handler: Callable[['Connection', IpcBaseEvent], None]):
+    def off(self, handler: Handler):
         """Unsubscribe the handler from being called on ipc events.
 
         :param handler: The handler that was previously attached with
@@ -389,9 +389,7 @@ def off(self, handler: Callable[['Connection', IpcBaseEvent], None]):
         """
         self._pubsub.unsubscribe(handler)
 
-    def on(self,
-           event: Union[Event, str],
-           handler: Optional[Callable[['Connection', IpcBaseEvent], None]] = None):
+    def on(self, event: Union[Event, str], handler: Optional[Handler] = None):
         def on_wrapped(handler):
             self._on(event, handler)
             return handler
@@ -401,7 +399,7 @@ def on_wrapped(handler):
         else:
             return on_wrapped
 
-    def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBaseEvent], None]):
+    def _on(self, event: Union[Event, str], handler: Handler):
         """Subscribe to the event and call the handler when it is emitted by
         the i3 ipc.