Type Annotations

[jfhbrook]

I'm working on adding type annotations to pyee in this branch.

It turns out that EventEmitters are not particularly type-safe, so the types are pretty hairy. The base is checking though! I think this is a good sign.

[jfhbrook]

It's tempting to break this up into multiple data structures with improved type safety:

self._user_events: Dict[UserEvent, "OrderedDict[HandlerP[Arg, Kwarg]. HamdlerP[Arg, Kwarg]]"]
self._error_events: "OrderedDict[HandlerP{Exception, None]], HandlerP[Exception, None]]"
self._listener_events: "OrderedDict[[HandlerP[Union[UserEvent, ErrorEvent, NewListenerEvent], HandlerP[Any, Any]]"

I'm concerned that I won't be able to narrow the type coming in from the type-unsafe on method, so I'll probably need to do something like:

self._all_events: Dict[Union[UserEvent, ErrorEvent, NewListenerEvent], "OrderedDict[HandlerP[Any, Any]]"]

But the reverse problem is true as well - emit_error would need to take a Handler[Any, Any] if pulling from self._all_events and convince pyright that it's a Handler[Exception, None]. One thing I can do is maintain all four data structures. I could also potentially make a wrapper type:

class InternalHandlerP(Generic[Event, Arg, Kwarg]):
    @property
    def event(self) -> Event: ...

    @property
    def handler(self) -> HandlerP[Arg, Kwarg]: ...

class UserHandler(InternalHandler[Event, Arg, Kwarg]):
    def __init__(self, event: UserEvent, handler: HandlerP[Arg, Kwarg]): ...
        self.event: UserEvent = event
        self.handler: HandlerP[Arg, Kwarg] = handler

class ErrorHandler(InternalHandlerP[ErrorEvent, HandlerP[Exception, None]):
    def __init__(self, handler: HandlerP[Exception, None]):
        self.event: ErrorEvent = "error"
        self.handler: HandlerP[Exception, None]

class NewListenerHandler(InternalHandlerP[NewListenerEvent, HandlerP[Union[UserEvent, ErrorEvent, NewListenerEvent, HandlerP[Any, Any], None]):
    def __init__(self, handler: HandlerP[Exception, None]):
        self.event: NewListenerEvent = "new_listener"
        self.handler: HandlerP[Union[UserEvent, ErrorEvent, NewListenerEvent, HandlerP[Any, Any], None] = handler

...

self._events: Dict[Union[Event, ErrorEvent, NewListenerEvent], "OrderedDict[HandlerP[Any, Any], InternalHandlerP]"]

This way, I should be able to do something like:

for handler in ist(self._events[event].values()):
    if handler.event == "error":
        # We should be able to infer the type of handler.handler now
    elif handler.event == "new_listener":
        # Same
    else:
        # Should be a HandlerP[Arg, Kwarg]

I think this will work!

[jfhbrook]

These data structures didn't work out with the existing abstractions, but it's definitely a step closer and I should reference this later.

[jfhbrook]

After a lot of thought, I settled on using str and Callable (unparametrized) for the existing abstractions. It doesn't afford much type safety, but it's what's actually possible to type consistently and sensibly. It simply turns out that these abstractions aren't type-safe! However, I think I can implement an alternate pyee.typed collection of EventEmitters, which make the necessary changes to support typed use.

There are a number of major issues with annotating the current implementation. To start, ee.on returns a union type, something like this:

Handler = Callable

def on(self, event: string,f: Optional[Handler]) -> Union[Handler, Callable[[Handler], Handler]]: ...

Even in the case where Handler isn't a parameterized type, this is still pretty unwieldy. This is pretty easy to address by using two functions:

def on(self, event: string, f: Handler) -> Handler: ...

def listener(self, event: string) -> Callable[[Handler], Handler]: ...

The second problem comes up when we try to parameterize the Handler. Right now it's any callable. But we want to use types to ensure that any handler added with a call to ee.on (or @ee.listener) has the correct signature. Right now, the event handlers are called with *args and **kwargs. These are brutal for typing because there can be any number of them and they all need to be encompassed by only two types. The upshot is that we can at least write a protocol for it:

A = TypeVar(name="A", contravariant=True)
K = TypeVar(name="K", contravariant=True)

class UserHandlerP(Protocol[A, K])):
    def __call__(self, *args: A, **kwargs: K) -> Any: ...

The handler is fire-and-forget, so there's no particular need to know or care about the return value. In a typed version it may return None as a convention for side effects, but in the case of the existing EventEmitter this is the most compassionate thing.

However, there are two internal events (three, in the case of TwistedEventEmitter): "error", and "new_listener".

One interesting thing we can do is create types for them:

ErrorEvent = Literal["error"]
NewListenerEvent = Literal["new_listener"]

and use a Union type to allow custom user event types:

def emit(self, event: Union[E, ErrorEvent, NewListenerEvent]) -> None

but notice that the type signature for the error handler isn't the same as the user handler:

class ErrorHandler(Protocol[E]):
    def __call__(self, error: E) -> Any: ...

and then we look at the "new_listener" handler. Its type looks something like this:

UserHandler = Callable
ErrorHandler = Callable

class NewListenerHandlerP(Protocol[E, A, K]):
    def __call__(
        self,
        event: Union[E, ErrorEvent, NewListenerEvent],
        handler: Union[
            HandlerP[A, K],
            ErrorHandlerP,
            "NewListenerHandlerP[E, A, K]"
        ]
    ]) -> Any

Another incredibly unwieldy type.You can imagine what would happen if you tried to add parameterized error handlers analogs to E, A and K - important for supporting the TwistedEventEmitter's "failure" event (ie, NewListenerHandler[Ev, ErrEv, A, K, Err]).

But all of those are surmountable if you're ok lugging around these complex types. The coup de grace has to do with the way event handlers are stored and retrieved. The way events are stored internally looks something like this:

event_table: Dict[Event, "t[OrderedDict[Handler, Handler]"] = defaultdict(OrderedDict)

The key handler is used for lookups when removing listeners, and the value is what actually gets called. These may differ to support use cases such as once, where the handler removes itself from the event table.

The problem is that the Handler type encompasses all three of our handlers, which each have their own signatures:

Handler = Union[UserHandler, ErrorHandler, AnyHandler]

and so when we fetch the handler on an emit, we no longer know which type of handler it is:

for f in self.events[event].values():
    # f: Union[UserHandler, ErrorHandler, AnyHandler]

Typically when this happens, what you can do is use a type guard to recover the type. This implies that you either have a fallback plan or intend to throw an error if it's not what you think it should be. For example, if my handlers were classes:

from pyee.typed import Handler, ErrorHandler

class MyHandler(Handler):
    def __call__(self, *args, Any, **kwargs: Any) -> Any:
        print(args, kwargs)

class MyErrorHandler(ErrorHandler[MyError][):
    def __call__(self, err: MyError) -> Any:
        ...

I could disambiguate them like so:

if isinstance(t, Handler):
    f(*args, **kwargs)
else:
    # Maybe it's an ErrorHandler on accident?
    raise Exception(f"unexpected handler: {f}")
...

you can see that this shows some risk - you have to be careful to pull out the right type for the right event. This gets worse if you try to associate certain types with certain events. You can use wrapper types that carry the type around, something like:

class Handler(Generic[T]):
    event_type: T
    __call__(self, data: T) -> Any: ...
        ...

you could recover the T with if isinstance(event, handler.event_type).

But either way this all depends on being able to effective guard for the involved types - and this is the coup de grace for these ideas. You can handle literals with if event == "error": and you can handle nominal types (if isinstance(event, handler.event_type): but you can't tell the difference between two functions via the type system.Once it's in the table, it can't come back out.

So that's basically dead in the water. But it should be pretty easy to make a new pyee.typed module which introduces type safety. If we make the following changes:

1. All handlers, regardless of type, take exactly one argument, which must either be a class or a string (which may be literally typed).
2. All event objects must be classes, such that their value may be used to guard after event lookup.

So maybe:

@dataclass
class Data:
    message: str


ee: TypedEventEmitter[Data] = TypedEventEmitter()


@ee.listens_to(Data)
def data_handler(data: Data):
    print(data.message)


ee.emit(Event(message="hello TypedEventEmitter!"))

Anything that can guard on instance checks should be good to go. In theory, I should be able to handle internal events pretty directly:

@ee.listens_to(Exception)
def on_error(exc: Exception):
    raise exc

New listener handling can be handled with an internal wrapper type:

E = TypeVar(name="E")
H = TypeVar(name="H")

@dataclass
class NewListener(Generic[E]):
    event:E,
    handler: Callable[[E], None]

I feel pretty good about this sketch, but it will need to evolve as I learn more during implementation.

[jfhbrook]

As it stands, I think I want to close this off and cut a release - it looks like I forgot to do a release recently, so I'll need to collate everything and get this out there.