"""Type classes for :mod:`mode.signals`."""
import abc
import asyncio
import typing
from weakref import ReferenceType
from typing import (
Any,
Awaitable,
Callable,
Generic,
MutableMapping,
MutableSet,
Optional,
Type,
TypeVar,
Union,
)
from mypy_extensions import KwArg, NamedArg, VarArg
__all__ = [
'BaseSignalT',
'FilterReceiverMapping',
'SignalHandlerT',
'SignalHandlerRefT',
'SignalT',
'SyncSignalT',
'T',
'T_contra',
]
T = TypeVar('T')
T_contra = TypeVar('T_contra', contravariant=True)
SignalHandlerT = Union[
Callable[
[T, VarArg(), NamedArg('BaseSignalT', name='signal'), KwArg()],
None,
],
Callable[
[T, VarArg(), NamedArg('BaseSignalT', name='signal'), KwArg()],
Awaitable[None],
],
]
if typing.TYPE_CHECKING:
SignalHandlerRefT = Union[
Callable[[], SignalHandlerT],
ReferenceType[SignalHandlerT]]
else:
SignalHandlerRefT = Any
FilterReceiverMapping = MutableMapping[Any, MutableSet[SignalHandlerRefT]]
[docs]class BaseSignalT(Generic[T]):
"""Base type for all signals."""
name: str
owner: Optional[Type]
@abc.abstractmethod
def __init__(self, *,
name: str = None,
owner: Type = None,
loop: asyncio.AbstractEventLoop = None,
default_sender: Any = None,
receivers: MutableSet[SignalHandlerRefT] = None,
filter_receivers: FilterReceiverMapping = None) -> None:
...
[docs] @abc.abstractmethod
def clone(self, **kwargs: Any) -> 'BaseSignalT':
...
[docs] @abc.abstractmethod
def with_default_sender(self, sender: Any = None) -> 'BaseSignalT':
...
[docs] @abc.abstractmethod
def connect(self, fun: SignalHandlerT, **kwargs: Any) -> Callable:
...
[docs] @abc.abstractmethod
def disconnect(self, fun: SignalHandlerT,
*,
sender: Any = None,
weak: bool = True) -> None:
...
[docs]class SignalT(BaseSignalT[T]):
"""Base class for all async signals (using ``async def``)."""
@abc.abstractmethod
async def __call__(self, sender: T_contra,
*args: Any, **kwargs: Any) -> None:
...
@abc.abstractmethod
async def send(self, sender: T_contra, *args: Any, **kwargs: Any) -> None:
...
[docs] @typing.no_type_check
@abc.abstractmethod
def clone(self, **kwargs: Any) -> 'SignalT':
...
[docs] @typing.no_type_check
@abc.abstractmethod
def with_default_sender(self, sender: Any = None) -> 'SignalT':
...
[docs]class SyncSignalT(BaseSignalT[T]):
"""Base class for all synchronous signals (using regular ``def``)."""
@abc.abstractmethod
def __call__(self, sender: T_contra, *args: Any, **kwargs: Any) -> None:
...
[docs] @abc.abstractmethod
def send(self, sender: T_contra, *args: Any, **kwargs: Any) -> None:
...
[docs] @typing.no_type_check
@abc.abstractmethod
def clone(self, **kwargs: Any) -> 'SyncSignalT':
...
[docs] @typing.no_type_check
@abc.abstractmethod
def with_default_sender(self, sender: Any = None) -> 'SyncSignalT':
...
