Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiosignal/__init__.py: 61%

1import sys 

2from typing import Any, Awaitable, Callable, TypeVar 

3 

4from frozenlist import FrozenList 

5 

6if sys.version_info >= (3, 11): 

7 from typing import Unpack 

8else: 

9 from typing_extensions import Unpack 

10 

11if sys.version_info >= (3, 13): 

12 from typing import TypeVarTuple 

13else: 

14 from typing_extensions import TypeVarTuple 

15 

16_T = TypeVar("_T") 

17_Ts = TypeVarTuple("_Ts", default=Unpack[tuple[()]]) 

18 

19__version__ = "1.4.0" 

20 

21__all__ = ("Signal",) 

22 

23 

24class Signal(FrozenList[Callable[[Unpack[_Ts]], Awaitable[object]]]): 

25 """Coroutine-based signal implementation. 

26 

27 To connect a callback to a signal, use any list method. 

28 

29 Signals are fired using the send() coroutine, which takes named 

30 arguments. 

31 """ 

32 

33 __slots__ = ("_owner",) 

34 

35 def __init__(self, owner: object): 

36 super().__init__() 

37 self._owner = owner 

38 

39 def __repr__(self) -> str: 

40 return "<Signal owner={}, frozen={}, {!r}>".format( 

41 self._owner, self.frozen, list(self) 

42 ) 

43 

44 async def send(self, *args: Unpack[_Ts], **kwargs: Any) -> None: 

45 """ 

46 Sends data to all registered receivers. 

47 """ 

48 if not self.frozen: 

49 raise RuntimeError("Cannot send non-frozen signal.") 

50 

51 for receiver in self: 

52 await receiver(*args, **kwargs) 

53 

54 def __call__( 

55 self, func: Callable[[Unpack[_Ts]], Awaitable[_T]] 

56 ) -> Callable[[Unpack[_Ts]], Awaitable[_T]]: 

57 """Decorator to add a function to this Signal.""" 

58 self.append(func) 

59 return func