Add Self to typing

Author: jfhbrook

pyee/__init__.py

@@ -27,6 +27,6 @@
 
 """
 
-from pyee.base import EventEmitter, PyeeError, PyeeException
+from pyee.base import EventEmitter, Handler, PyeeError, PyeeException
 
-__all__ = ["EventEmitter", "PyeeError", "PyeeException"]
+__all__ = ["EventEmitter", "Handler", "PyeeError", "PyeeException"]

pyee/asyncio.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 from asyncio import AbstractEventLoop, ensure_future, Future, iscoroutine, wait
-from typing import Any, Callable, cast, Dict, Optional, Set, Tuple
+from typing import Any, Callable, cast, Dict, Optional, Self, Set, Tuple
 
 from pyee.base import EventEmitter
 
@@ -36,13 +36,13 @@ async def async_handler(*args, **kwargs):
     coroutine is scheduled in a fire-and-forget fashion.
     """
 
-    def __init__(self, loop: Optional[AbstractEventLoop] = None):
+    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: Self,
         event: str,
         *args: Any,
         **kwargs: Any,
@@ -68,11 +68,11 @@ def emit(
         return super().emit(event, *args, **kwargs)
 
     def _emit_run(
-        self,
+        self: Self,
         f: Callable,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
-    ):
+    ) -> None:
         try:
             coro: Any = f(*args, **kwargs)
         except Exception as exc:
@@ -105,7 +105,7 @@ def callback(f):
             fut.add_done_callback(callback)
             self._waiting.add(fut)
 
-    async def wait_for_complete(self):
+    async def wait_for_complete(self: Self) -> None:
         """Waits for all pending tasks to complete. For example:
 
         ```py
@@ -128,7 +128,7 @@ async def async_handler(*args, **kwargs):
         if self._waiting:
             await wait(self._waiting)
 
-    def cancel(self):
+    def cancel(self: Self) -> None:
         """Cancel all pending tasks. For example:
 
         ```py
@@ -153,7 +153,7 @@ async def async_handler(*args, **kwargs):
         self._waiting.clear()
 
     @property
-    def complete(self) -> bool:
+    def complete(self: Self) -> bool:
         """When true, there are no pending tasks, and execution is complete.
         For example:
 

pyee/base.py

@@ -9,6 +9,7 @@
     List,
     Mapping,
     Optional,
+    Self,
     Set,
     Tuple,
     TypeVar,
@@ -56,24 +57,24 @@ def on_error(message):
     your own exceptions, and treat them accordingly.
     """
 
-    def __init__(self) -> None:
+    def __init__(self: Self) -> None:
         self._events: Dict[
             str,
             "OrderedDict[Callable, Callable]",
         ] = dict()
         self._lock: Lock = Lock()
 
-    def __getstate__(self) -> Mapping[str, Any]:
+    def __getstate__(self: Self) -> Mapping[str, Any]:
         state = self.__dict__.copy()
         del state["_lock"]
         return state
 
-    def __setstate__(self, state: Mapping[str, Any]) -> None:
+    def __setstate__(self: Self, state: Mapping[str, Any]) -> None:
         self.__dict__.update(state)
         self._lock = Lock()
 
     def on(
-        self, event: str, f: Optional[Handler] = None
+        self: Self, event: str, f: Optional[Handler] = None
     ) -> Union[Handler, Callable[[Handler], Handler]]:
         """Registers the function `f` to the event name `event`, if provided.
 
@@ -106,7 +107,7 @@ def data_handler(data):
         else:
             return self.add_listener(event, f)
 
-    def listens_to(self, event: str) -> Callable[[Handler], Handler]:
+    def listens_to(self: Self, event: str) -> Callable[[Handler], Handler]:
         """Returns a decorator which will register the decorated function to
         the event name `event`:
 
@@ -126,7 +127,7 @@ def on(f: Handler) -> Handler:
 
         return on
 
-    def add_listener(self, event: str, f: Handler) -> Handler:
+    def add_listener(self: Self, event: str, f: Handler) -> Handler:
         """Register the function `f` to the event name `event`:
 
         ```
@@ -142,7 +143,7 @@ def data_handler(data):
         self._add_event_handler(event, f, f)
         return f
 
-    def _add_event_handler(self, event: str, k: Callable, v: Callable):
+    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)
 
@@ -156,26 +157,26 @@ def _add_event_handler(self, event: str, k: Callable, v: Callable):
             self._events[event][k] = v
 
     def _emit_run(
-        self,
+        self: Self,
         f: Callable,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
     ) -> None:
         f(*args, **kwargs)
 
-    def event_names(self) -> Set[str]:
+    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, event: str, error: Any) -> None:
+    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: Self,
         event: str,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
@@ -191,7 +192,7 @@ def _call_handlers(
         return handled
 
     def emit(
-        self,
+        self: Self,
         event: str,
         *args: Any,
         **kwargs: Any,
@@ -217,7 +218,7 @@ def emit(
         return handled
 
     def once(
-        self,
+        self: Self,
         event: str,
         f: Optional[Callable] = None,
     ) -> Callable:
@@ -249,18 +250,18 @@ def g(
         else:
             return _wrapper(f)
 
-    def _remove_listener(self, event: str, f: Callable) -> None:
+    def _remove_listener(self: Self, event: str, f: Callable) -> None:
         """Naked unprotected removal."""
         self._events[event].pop(f)
         if not len(self._events[event]):
             del self._events[event]
 
-    def remove_listener(self, event: str, f: Callable) -> None:
+    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, event: Optional[str] = None) -> None:
+    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.
         """
@@ -270,6 +271,6 @@ def remove_all_listeners(self, event: Optional[str] = None) -> None:
             else:
                 self._events = dict()
 
-    def listeners(self, event: str) -> List[Callable]:
+    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())

pyee/cls.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from functools import wraps
-from typing import Callable, List, Type, TypeVar
+from typing import Any, Callable, Iterator, List, Type, TypeVar
 
 from pyee import EventEmitter
 
@@ -12,13 +12,13 @@ class Handler:
 
 
 class Handlers:
-    def __init__(self):
+    def __init__(self) -> None:
         self._handlers: List[Handler] = []
 
-    def append(self, handler):
+    def append(self, handler) -> None:
         self._handlers.append(handler)
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[Handler]:
         return iter(self._handlers)
 
     def reset(self):
@@ -41,9 +41,9 @@ def decorator(method: Callable) -> Callable:
     return decorator
 
 
-def _bind(self, method):
+def _bind(self: Any, method: Any) -> Any:
     @wraps(method)
-    def bound(*args, **kwargs):
+    def bound(*args, **kwargs) -> Any:
         return method(self, *args, **kwargs)
 
     return bound
@@ -103,7 +103,7 @@ async def event_handler(self, *args, **kwargs):
     og_init: Callable = cls.__init__
 
     @wraps(cls.__init__)
-    def init(self, *args, **kwargs):
+    def init(self: Any, *args: Any, **kwargs: Any) -> None:
         og_init(self, *args, **kwargs)
         if not hasattr(self, "event_emitter"):
             self.event_emitter = EventEmitter()

pyee/executor.py

@@ -2,7 +2,7 @@
 
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
 from types import TracebackType
-from typing import Any, Callable, Dict, Optional, Tuple, Type
+from typing import Any, Callable, Dict, Optional, Self, Tuple, Type
 
 from pyee.base import EventEmitter
 
@@ -44,19 +44,19 @@ def handler(data):
     No effort is made to ensure thread safety, beyond using an executor.
     """
 
-    def __init__(self, executor: Optional[Executor] = None):
+    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: Self,
         f: Callable,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
-    ):
+    ) -> None:
         future: Future = self._executor.submit(f, *args, **kwargs)
 
         @future.add_done_callback
@@ -67,15 +67,15 @@ def _callback(f: Future) -> None:
             elif exc is not None:
                 raise exc
 
-    def shutdown(self, wait: bool = True) -> None:
+    def shutdown(self: Self, wait: bool = True) -> None:
         """Call `shutdown` on the internal executor."""
 
         self._executor.shutdown(wait=wait)
 
-    def __enter__(self) -> "ExecutorEventEmitter":
+    def __enter__(self: Self) -> "ExecutorEventEmitter":
         return self
 
     def __exit__(
-        self, type: Type[Exception], value: Exception, traceback: TracebackType
+        self: Self, type: Type[Exception], value: Exception, traceback: TracebackType
     ) -> Optional[bool]:
         self.shutdown()

pyee/trio.py

@@ -2,7 +2,17 @@
 
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from types import TracebackType
-from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional, Tuple, Type
+from typing import (
+    Any,
+    AsyncGenerator,
+    Awaitable,
+    Callable,
+    Dict,
+    Optional,
+    Self,
+    Tuple,
+    Type,
+)
 
 import trio
 
@@ -50,7 +60,7 @@ async def handler(data):
     """
 
     def __init__(
-        self,
+        self: Self,
         nursery: Optional[Nursery] = None,
         manager: Optional["AbstractAsyncContextManager[trio.Nursery]"] = None,
     ):
@@ -69,7 +79,7 @@ def __init__(
             self._manager = trio.open_nursery()
 
     def _async_runner(
-        self,
+        self: Self,
         f: Callable,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
@@ -83,7 +93,7 @@ async def runner() -> None:
         return runner
 
     def _emit_run(
-        self,
+        self: Self,
         f: Callable,
         args: Tuple[Any, ...],
         kwargs: Dict[str, Any],
@@ -94,7 +104,7 @@ def _emit_run(
 
     @asynccontextmanager
     async def context(
-        self,
+        self: Self,
     ) -> AsyncGenerator["TrioEventEmitter", None]:
         """Returns an async contextmanager which manages the underlying
         nursery to the EventEmitter. The `TrioEventEmitter`'s
@@ -110,14 +120,14 @@ async def context(
         else:
             raise PyeeError("Uninitialized nursery or nursery manager")
 
-    async def __aenter__(self) -> "TrioEventEmitter":
+    async def __aenter__(self: Self) -> "TrioEventEmitter":
         self._context: Optional[AbstractAsyncContextManager["TrioEventEmitter"]] = (
             self.context()
         )
         return await self._context.__aenter__()
 
     async def __aexit__(
-        self,
+        self: Self,
         type: Optional[Type[BaseException]],
         value: Optional[BaseException],
         traceback: Optional[TracebackType],