API Docs

class EventEmitter:
    """The base event emitter class. All other event emitters inherit from
    this class.

    Most events are registered with an emitter via the `on` and `once`
    methods, and fired with the `emit` method. However, pyee event emitters
    have two *special* events:

    - `new_listener`: Fires whenever a new listener is created. Listeners for
      this event do not fire upon their own creation.

    - `error`: When emitted raises an Exception by default, behavior can be
      overridden by attaching callback to the event.

      For example:

    ```py
    @ee.on('error')
    def on_error(message):
        logging.err(message)

    ee.emit('error', Exception('something blew up'))
    ```

    All callbacks are handled in a synchronous, blocking manner. As in node.js,
    raised exceptions are not automatically handled for you---you must catch
    your own exceptions, and treat them accordingly.
    """

    def __init__(self: Self) -> None:
        self._events: Dict[
            str,
            "OrderedDict[Callable, Callable]",
        ] = dict()
        self._lock: Lock = Lock()

    def __getstate__(self: Self) -> Mapping[str, Any]:
        state = self.__dict__.copy()
        del state["_lock"]
        return state

    def __setstate__(self: Self, state: Mapping[str, Any]) -> None:
        self.__dict__.update(state)
        self._lock = Lock()

    @overload
    def on(self: Self, event: str) -> Callable[[Handler], Handler]: ...
    @overload
    def on(self: Self, event: str, f: Handler) -> Handler: ...

    def on(
        self: Self, event: str, f: Optional[Handler] = None
    ) -> Union[Handler, Callable[[Handler], Handler]]:
        """Registers the function `f` to the event name `event`, if provided.

        If `f` isn't provided, this method calls `EventEmitter#listens_to`, and
        otherwise calls `EventEmitter#add_listener`. In other words, you may either
        use it as a decorator:

        ```py
        @ee.on('data')
        def data_handler(data):
            print(data)
        ```

        Or directly:

        ```py
        ee.on('data', data_handler)
        ```

        In both the decorated and undecorated forms, the event handler is
        returned. The upshot of this is that you can call decorated handlers
        directly, as well as use them in remove_listener calls.

        Note that this method's return type is a union type. If you are using
        mypy or pyright, you will probably want to use either
        `EventEmitter#listens_to` or `EventEmitter#add_listener`.
        """
        if f is None:
            return self.listens_to(event)
        else:
            return self.add_listener(event, f)

    def listens_to(self: Self, event: str) -> Callable[[Handler], Handler]:
        """Returns a decorator which will register the decorated function to
        the event name `event`:

        ```py
        @ee.listens_to("event")
        def data_handler(data):
            print(data)
        ```

        By only supporting the decorator use case, this method has improved
        type safety over `EventEmitter#on`.
        """

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

        return on

    def add_listener(self: Self, event: str, f: Handler) -> Handler:
        """Register the function `f` to the event name `event`:

        ```
        def data_handler(data):
            print(data)

        h = ee.add_listener("event", data_handler)
        ```

        By not supporting the decorator use case, this method has improved
        type safety over `EventEmitter#on`.
        """
        self._add_event_handler(event, f, f)
        return f

    def _add_event_handler(self: Self, event: str, k: Callable, v: Callable):
        # Fire 'new_listener' *before* adding the new listener!
        self.emit("new_listener", event, k)

        # Add the necessary function
        # Note that k and v are the same for `on` handlers, but
        # different for `once` handlers, where v is a wrapped version
        # of k which removes itself before calling k
        with self._lock:
            if event not in self._events:
                self._events[event] = OrderedDict()
            self._events[event][k] = v

    def _emit_run(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        f(*args, **kwargs)

    def event_names(self: Self) -> Set[str]:
        """Get a set of events that this emitter is listening to."""
        return set(self._events.keys())

    def _emit_handle_potential_error(self: Self, event: str, error: Any) -> None:
        if event == "error":
            if isinstance(error, Exception):
                raise error
            else:
                raise PyeeError(f"Uncaught, unspecified 'error' event: {error}")

    def _call_handlers(
        self: Self,
        event: str,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> bool:
        handled = False

        with self._lock:
            funcs = list(self._events.get(event, OrderedDict()).values())
        for f in funcs:
            self._emit_run(f, args, kwargs)
            handled = True

        return handled

    def emit(
        self: Self,
        event: str,
        *args: Any,
        **kwargs: Any,
    ) -> bool:
        """Emit `event`, passing `*args` and `**kwargs` to each attached
        function. Returns `True` if any functions are attached to `event`;
        otherwise returns `False`.

        Example:

        ```py
        ee.emit('data', '00101001')
        ```

        Assuming `data` is an attached function, this will call
        `data('00101001')'`.
        """
        handled = self._call_handlers(event, args, kwargs)

        if not handled:
            self._emit_handle_potential_error(event, args[0] if args else None)

        return handled

    def once(
        self: Self,
        event: str,
        f: Optional[Callable] = None,
    ) -> Callable:
        """The same as `ee.on`, except that the listener is automatically
        removed after being called.
        """

        def _wrapper(f: Callable) -> Callable:
            def g(
                *args: Any,
                **kwargs: Any,
            ) -> Any:
                with self._lock:
                    # Check that the event wasn't removed already right
                    # before the lock
                    if event in self._events and f in self._events[event]:
                        self._remove_listener(event, f)
                    else:
                        return None
                # f may return a coroutine, so we need to return that
                # result here so that emit can schedule it
                return f(*args, **kwargs)

            self._add_event_handler(event, f, g)
            return f

        if f is None:
            return _wrapper
        else:
            return _wrapper(f)

    def _remove_listener(self: Self, event: str, f: Callable) -> None:
        """Naked unprotected removal."""
        if event in self._events:
            self._events[event].pop(f)
            if not self._events[event]:
                del self._events[event]

    def remove_listener(self: Self, event: str, f: Callable) -> None:
        """Removes the function `f` from `event`."""
        with self._lock:
            self._remove_listener(event, f)

    def remove_all_listeners(self: Self, event: Optional[str] = None) -> None:
        """Remove all listeners attached to `event`.
        If `event` is `None`, remove all listeners on all events.
        """
        with self._lock:
            if event is not None:
                self._events[event] = OrderedDict()
            else:
                self._events = dict()

    def listeners(self: Self, event: str) -> List[Callable]:
        """Returns a list of all listeners registered to the `event`."""
        return list(self._events.get(event, OrderedDict()).keys())

class PyeeException(Exception):
    """An exception internal to pyee. Deprecated in favor of PyeeError."""

class AsyncIOEventEmitter(EventEmitter):
    """An event emitter class which can run asyncio coroutines in addition to
    synchronous blocking functions. For example:

    ```py
    @ee.on('event')
    async def async_handler(*args, **kwargs):
        await returns_a_future()
    ```

    On emit, the event emitter  will automatically schedule the coroutine using
    `asyncio.ensure_future` and the configured event loop (defaults to
    `asyncio.get_event_loop()`).

    Unlike the case with the EventEmitter, all exceptions raised by
    event handlers are automatically emitted on the `error` event. This is
    important for asyncio coroutines specifically but is also handled for
    synchronous functions for consistency.

    When `loop` is specified, the supplied event loop will be used when
    scheduling work with `ensure_future`. Otherwise, the default asyncio
    event loop is used.

    For asyncio coroutine event handlers, calling emit is non-blocking.
    In other words, you do not have to await any results from emit, and the
    coroutine is scheduled in a fire-and-forget fashion.
    """

    def __init__(self: Self, loop: Optional[AbstractEventLoop] = None) -> None:
        super(AsyncIOEventEmitter, self).__init__()
        self._loop: Optional[AbstractEventLoop] = loop
        self._waiting: Set[Future] = set()

    def emit(
        self: Self,
        event: str,
        *args: Any,
        **kwargs: Any,
    ) -> bool:
        """Emit `event`, passing `*args` and `**kwargs` to each attached
        function or coroutine. Returns `True` if any functions are attached to
        `event`; otherwise returns `False`.

        Example:

        ```py
        ee.emit('data', '00101001')
        ```

        Assuming `data` is an attached function, this will call
        `data('00101001')'`.

        When executing coroutine handlers, their respective futures will be
        stored in a "waiting" state. These futures may be waited on or
        canceled with `wait_for_complete` and `cancel`, respectively; and
        their status may be checked via the `complete` property.
        """
        return super().emit(event, *args, **kwargs)

    def _emit_run(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        try:
            coro: Any = f(*args, **kwargs)
        except Exception as exc:
            self.emit("error", exc)
        else:
            if iscoroutine(coro):
                if self._loop:
                    # ensure_future is *extremely* cranky about the types here,
                    # but this is relatively well-tested and I think the types
                    # are more strict than they should be
                    fut: Any = ensure_future(cast(Any, coro), loop=self._loop)
                else:
                    fut = ensure_future(cast(Any, coro))

            elif isinstance(coro, Future):
                fut = cast(Any, coro)
            else:
                return

            def callback(f: Future) -> None:
                self._waiting.discard(f)

                if f.cancelled():
                    return

                exc: Optional[BaseException] = f.exception()
                if exc:
                    self.emit("error", exc)

            fut.add_done_callback(callback)
            self._waiting.add(fut)

    async def wait_for_complete(self: Self) -> None:
        """Waits for all pending tasks to complete. For example:

        ```py
        @ee.on('event')
        async def async_handler(*args, **kwargs):
            await returns_a_future()

        # Triggers execution of async_handler
        ee.emit('data', '00101001')

        await ee.wait_for_complete()

        # async_handler has completed execution
        ```

        This is useful if you're attempting a graceful shutdown of your
        application and want to ensure all coroutines have completed execution
        beforehand.
        """
        if self._waiting:
            await wait(self._waiting)

    def cancel(self: Self) -> None:
        """Cancel all pending tasks. For example:

        ```py
        @ee.on('event')
        async def async_handler(*args, **kwargs):
            await returns_a_future()

        # Triggers execution of async_handler
        ee.emit('data', '00101001')

        ee.cancel()

        # async_handler execution has been canceled
        ```

        This is useful if you're attempting to shut down your application and
        attempts at a graceful shutdown via `wait_for_complete` have failed.
        """
        for fut in self._waiting:
            if not fut.done() and not fut.cancelled():
                fut.cancel()
        self._waiting.clear()

    @property
    def complete(self: Self) -> bool:
        """When true, there are no pending tasks, and execution is complete.
        For example:

        ```py
        @ee.on('event')
        async def async_handler(*args, **kwargs):
            await returns_a_future()

        # Triggers execution of async_handler
        ee.emit('data', '00101001')

        # async_handler is still running, so this prints False
        print(ee.complete)

        await ee.wait_for_complete()

        # async_handler has completed execution, so this prints True
        print(ee.complete)
        ```
        """
        return not self._waiting

class TwistedEventEmitter(EventEmitter):
    """An event emitter class which can run twisted coroutines and handle
    returned Deferreds, in addition to synchronous blocking functions. For
    example:

    ```py
    @ee.on('event')
    @inlineCallbacks
    def async_handler(*args, **kwargs):
        yield returns_a_deferred()
    ```

    or:

    ```py
    @ee.on('event')
    async def async_handler(*args, **kwargs):
        await returns_a_deferred()
    ```


    When async handlers fail, Failures are first emitted on the `failure`
    event. If there are no `failure` handlers, the Failure's associated
    exception is then emitted on the `error` event. If there are no `error`
    handlers, the exception is raised. For consistency, when handlers raise
    errors synchronously, they're captured, wrapped in a Failure and treated
    as an async failure. This is unlike the behavior of EventEmitter,
    which have no special error handling.

    For twisted coroutine event handlers, calling emit is non-blocking.
    In other words, you do not have to await any results from emit, and the
    coroutine is scheduled in a fire-and-forget fashion.

    Similar behavior occurs for "sync" functions which return Deferreds.
    """

    def __init__(self: Self) -> None:
        super(TwistedEventEmitter, self).__init__()

    def _emit_run(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        d: Optional[Deferred[Any]] = None
        try:
            result = f(*args, **kwargs)
        except Exception:
            self.emit("failure", Failure())
        else:
            if iscoroutine(result):
                d = ensureDeferred(result)
            elif isinstance(result, Deferred):
                d = result
            elif not d:
                return

            def errback(failure: Failure) -> None:
                if failure:
                    self.emit("failure", failure)

            d.addErrback(errback)

    def _emit_handle_potential_error(self: Self, event: str, error: Any) -> None:
        if event == "failure":
            if isinstance(error, Failure):
                try:
                    error.raiseException()
                except Exception as exc:
                    self.emit("error", exc)
            elif isinstance(error, Exception):
                self.emit("error", error)
            else:
                self.emit("error", PyeeError(f"Unexpected failure object: {error}"))
        else:
            cast(Any, super(TwistedEventEmitter, self))._emit_handle_potential_error(
                event, error
            )

class ExecutorEventEmitter(EventEmitter):
    """An event emitter class which runs handlers in a `concurrent.futures`
    executor.

    By default, this class creates a default `ThreadPoolExecutor`, but
    a custom executor may also be passed in explicitly to, for instance,
    use a `ProcessPoolExecutor` instead.

    This class runs all emitted events on the configured executor. Errors
    captured by the resulting Future are automatically emitted on the
    `error` event. This is unlike the EventEmitter, which have no error
    handling.

    The underlying executor may be shut down by calling the `shutdown`
    method. Alternately you can treat the event emitter as a context manager:

    ```py
    with ExecutorEventEmitter() as ee:
        # Underlying executor open

        @ee.on('data')
        def handler(data):
            print(data)

        ee.emit('event')

    # Underlying executor closed
    ```

    Since the function call is scheduled on an executor, emit is always
    non-blocking.

    No effort is made to ensure thread safety, beyond using an executor.
    """

    def __init__(self: Self, executor: Optional[Executor] = None) -> None:
        super(ExecutorEventEmitter, self).__init__()
        if executor:
            self._executor: Executor = executor
        else:
            self._executor = ThreadPoolExecutor()

    def _emit_run(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        future: Future = self._executor.submit(f, *args, **kwargs)

        @future.add_done_callback
        def _callback(f: Future) -> None:
            exc: Optional[BaseException] = f.exception()
            if isinstance(exc, Exception):
                self.emit("error", exc)
            elif exc is not None:
                raise exc

    def shutdown(self: Self, wait: bool = True) -> None:
        """Call `shutdown` on the internal executor."""

        self._executor.shutdown(wait=wait)

    def __enter__(self: Self) -> "ExecutorEventEmitter":
        return self

    def __exit__(
        self: Self, type: Type[Exception], value: Exception, traceback: TracebackType
    ) -> Optional[bool]:
        self.shutdown()
        return None

class TrioEventEmitter(EventEmitter):
    """An event emitter class which can run trio tasks in a trio nursery.

    By default, this class will lazily create both a nursery manager (the
    object returned from `trio.open_nursery()` and a nursery (the object
    yielded by using the nursery manager as an async context manager). It is
    also possible to supply an existing nursery manager via the `manager`
    argument, or an existing nursery via the `nursery` argument.

    Instances of TrioEventEmitter are themselves async context managers, so
    that they may manage the lifecycle of the underlying trio nursery. For
    example, typical usage of this library may look something like this::

    ```py
    async with TrioEventEmitter() as ee:
        # Underlying nursery is instantiated and ready to go
        @ee.on('data')
        async def handler(data):
            print(data)

        ee.emit('event')

    # Underlying nursery and manager have been cleaned up
    ```

    Unlike the case with the EventEmitter, all exceptions raised by event
    handlers are automatically emitted on the `error` event. This is
    important for trio coroutines specifically but is also handled for
    synchronous functions for consistency.

    For trio coroutine event handlers, calling emit is non-blocking. In other
    words, you should not attempt to await emit; the coroutine is scheduled
    in a fire-and-forget fashion.
    """

    def __init__(
        self: Self,
        nursery: Optional[Nursery] = None,
        manager: Optional["AbstractAsyncContextManager[trio.Nursery]"] = None,
    ):
        super(TrioEventEmitter, self).__init__()
        self._nursery: Optional[Nursery] = None
        self._manager: Optional["AbstractAsyncContextManager[trio.Nursery]"] = None
        if nursery:
            if manager:
                raise PyeeError(
                    "You may either pass a nursery or a nursery manager " "but not both"
                )
            self._nursery = nursery
        elif manager:
            self._manager = manager
        else:
            self._manager = trio.open_nursery()

    def _async_runner(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> Callable[[], Awaitable[None]]:
        async def runner() -> None:
            try:
                await f(*args, **kwargs)
            except Exception as exc:
                self.emit("error", exc)

        return runner

    def _emit_run(
        self: Self,
        f: Callable,
        args: Tuple[Any, ...],
        kwargs: Dict[str, Any],
    ) -> None:
        if not self._nursery:
            raise PyeeError("Uninitialized trio nursery")
        self._nursery.start_soon(self._async_runner(f, args, kwargs))

    @asynccontextmanager
    async def context(
        self: Self,
    ) -> AsyncGenerator["TrioEventEmitter", None]:
        """Returns an async contextmanager which manages the underlying
        nursery to the EventEmitter. The `TrioEventEmitter`'s
        async context management methods are implemented using this
        function, but it may also be used directly for clarity.
        """
        if self._nursery is not None:
            yield self
        elif self._manager is not None:
            async with self._manager as nursery:
                self._nursery = nursery
                yield self
        else:
            raise PyeeError("Uninitialized nursery or nursery manager")

    async def __aenter__(self: Self) -> "TrioEventEmitter":
        self._context: Optional[AbstractAsyncContextManager["TrioEventEmitter"]] = (
            self.context()
        )
        return await cast(Any, self._context).__aenter__()

    async def __aexit__(
        self: Self,
        type: Optional[Type[BaseException]],
        value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> Optional[bool]:
        if self._context is None:
            raise PyeeError("Attempting to exit uninitialized context")
        rv = await self._context.__aexit__(type, value, traceback)
        self._context = None
        self._nursery = None
        self._manager = None
        return rv