Skip to content

signal

Signal

Signal that notifies all receiver functions. In ormar used by models to send pre_save, post_save etc. signals.

Source code in ormar/signals/signal.py
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class Signal:
    """
    Signal that notifies all receiver functions.
    In ormar used by models to send pre_save, post_save etc. signals.
    """

    def __init__(self) -> None:
        self._receivers: Dict[Union[int, Tuple[int, int]], Callable] = {}

    def connect(self, receiver: Callable) -> None:
        """
        Connects given receiver function to the signal.

        :raises SignalDefinitionError: if receiver is not callable
        or not accept **kwargs
        :param receiver: receiver function
        :type receiver: Callable
        """
        if not callable(receiver):
            raise SignalDefinitionError("Signal receivers must be callable.")
        if not callable_accepts_kwargs(receiver):
            raise SignalDefinitionError(
                "Signal receivers must accept **kwargs argument."
            )
        new_receiver_key = make_id(receiver)
        if new_receiver_key not in self._receivers:
            self._receivers[new_receiver_key] = receiver

    def disconnect(self, receiver: Callable) -> bool:
        """
        Removes the receiver function from the signal.

        :param receiver: receiver function
        :type receiver: Callable
        :return: flag if receiver was removed
        :rtype: bool
        """
        new_receiver_key = make_id(receiver)
        receiver_func: Union[Callable, None] = self._receivers.pop(
            new_receiver_key, None
        )
        return True if receiver_func is not None else False

    async def send(self, sender: Type["Model"], **kwargs: Any) -> None:
        """
        Notifies all receiver functions with given kwargs
        :param sender: model that sends the signal
        :type sender: Type["Model"]
        :param kwargs: arguments passed to receivers
        :type kwargs: Any
        """
        receivers = [
            receiver_func(sender=sender, **kwargs)
            for receiver_func in self._receivers.values()
        ]
        await asyncio.gather(*receivers)

connect(receiver)

Connects given receiver function to the signal.

Parameters:

Name Type Description Default
receiver Callable

receiver function

required

Raises:

Type Description
SignalDefinitionError

if receiver is not callable or not accept **kwargs

Source code in ormar/signals/signal.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def connect(self, receiver: Callable) -> None:
    """
    Connects given receiver function to the signal.

    :raises SignalDefinitionError: if receiver is not callable
    or not accept **kwargs
    :param receiver: receiver function
    :type receiver: Callable
    """
    if not callable(receiver):
        raise SignalDefinitionError("Signal receivers must be callable.")
    if not callable_accepts_kwargs(receiver):
        raise SignalDefinitionError(
            "Signal receivers must accept **kwargs argument."
        )
    new_receiver_key = make_id(receiver)
    if new_receiver_key not in self._receivers:
        self._receivers[new_receiver_key] = receiver

disconnect(receiver)

Removes the receiver function from the signal.

Parameters:

Name Type Description Default
receiver Callable

receiver function

required

Returns:

Type Description
bool

flag if receiver was removed

Source code in ormar/signals/signal.py
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def disconnect(self, receiver: Callable) -> bool:
    """
    Removes the receiver function from the signal.

    :param receiver: receiver function
    :type receiver: Callable
    :return: flag if receiver was removed
    :rtype: bool
    """
    new_receiver_key = make_id(receiver)
    receiver_func: Union[Callable, None] = self._receivers.pop(
        new_receiver_key, None
    )
    return True if receiver_func is not None else False

send(sender, **kwargs) async

Notifies all receiver functions with given kwargs

Parameters:

Name Type Description Default
sender Type['Model']

model that sends the signal

required
kwargs Any

arguments passed to receivers

required
Source code in ormar/signals/signal.py
84
85
86
87
88
89
90
91
92
93
94
95
96
async def send(self, sender: Type["Model"], **kwargs: Any) -> None:
    """
    Notifies all receiver functions with given kwargs
    :param sender: model that sends the signal
    :type sender: Type["Model"]
    :param kwargs: arguments passed to receivers
    :type kwargs: Any
    """
    receivers = [
        receiver_func(sender=sender, **kwargs)
        for receiver_func in self._receivers.values()
    ]
    await asyncio.gather(*receivers)

SignalEmitter

Bases: dict

Emitter that registers the signals in internal dictionary. If signal with given name does not exist it's auto added on access.

Source code in ormar/signals/signal.py
 99
100
101
102
103
104
105
106
107
108
109
110
111
class SignalEmitter(dict):
    """
    Emitter that registers the signals in internal dictionary.
    If signal with given name does not exist it's auto added on access.
    """

    def __getattr__(self, item: str) -> Signal:
        return self.setdefault(item, Signal())

    def __setattr__(self, key: str, value: Signal) -> None:
        if not isinstance(value, Signal):
            raise SignalDefinitionError(f"{value} is not valid signal")
        self[key] = value

callable_accepts_kwargs(func)

Checks if function accepts **kwargs.

Parameters:

Name Type Description Default
func Callable

function which signature needs to be checked

required

Returns:

Type Description
bool

result of the check

Source code in ormar/signals/signal.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def callable_accepts_kwargs(func: Callable) -> bool:
    """
    Checks if function accepts **kwargs.

    :param func: function which signature needs to be checked
    :type func: function
    :return: result of the check
    :rtype: bool
    """
    return any(
        p
        for p in inspect.signature(func).parameters.values()
        if p.kind == p.VAR_KEYWORD
    )

make_id(target)

Creates id of a function or method to be used as key to store signal

Parameters:

Name Type Description Default
target Any

target which id we want

required

Returns:

Type Description
int

id of the target

Source code in ormar/signals/signal.py
27
28
29
30
31
32
33
34
35
36
37
38
def make_id(target: Any) -> Union[int, Tuple[int, int]]:
    """
    Creates id of a function or method to be used as key to store signal

    :param target: target which id we want
    :type target: Any
    :return: id of the target
    :rtype: int
    """
    if hasattr(target, "__func__"):
        return id(target.__self__), id(target.__func__)
    return id(target)