Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def receivers_for(self, sender):
"""Iterate all live receivers listening for *sender*."""
# TODO: test receivers_for(ANY)
if self.receivers:
sender_id = hashable_identity(sender)
if sender_id in self._by_sender:
ids = (self._by_sender[ANY_ID] |
self._by_sender[sender_id])
else:
ids = self._by_sender[ANY_ID].copy()
for receiver_id in ids:
receiver = self.receivers.get(receiver_id)
if receiver is None:
continue
if isinstance(receiver, WeakTypes):
strong = receiver()
if strong is None:
self._disconnect(receiver_id, ANY_ID)
continue
receiver = strong
yield receiver
def disconnect(self, receiver, sender=ANY):
"""Disconnect *receiver* from this signal's events.
:param receiver: a previously :meth:`connected` callable
:param sender: a specific sender to disconnect from, or :obj:`ANY`
to disconnect from all senders. Defaults to ``ANY``.
"""
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
receiver_id = hashable_identity(receiver)
self._disconnect(receiver_id, sender_id)
if ('receiver_disconnected' in self.__dict__ and
self.receiver_disconnected.receivers):
self.receiver_disconnected.send(self,
receiver=receiver,
sender=sender)
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
# wire together a cleanup for weakref-able senders
try:
sender_ref = reference(sender, self._cleanup_sender)
sender_ref.sender_id = sender_id
except TypeError:
pass
else:
self._weak_senders.setdefault(sender_id, sender_ref)
del sender_ref
`sender=` as a single positional argument and any \*\*kwargs that
were provided to a call to :meth:`send`.
:param sender: Any object or :obj:`ANY`, defaults to ``ANY``.
Restricts notifications delivered to *receiver* to only those
:meth:`send` emissions sent by *sender*. If ``ANY``, the receiver
will always be notified. A *receiver* may be connected to
multiple *sender* values on the same Signal through multiple calls
to :meth:`connect`.
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
`sender=` as a single positional argument and any \*\*kwargs that
were provided to a call to :meth:`send`.
:param sender: Any object or :obj:`ANY`, defaults to ``ANY``.
Restricts notifications delivered to *receiver* to only those
:meth:`send` emissions sent by *sender*. If ``ANY``, the receiver
will always be notified. A *receiver* may be connected to
multiple *sender* values on the same Signal through multiple calls
to :meth:`connect`.
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
# wire together a cleanup for weakref-able senders
try:
sender_ref = reference(sender, self._cleanup_sender)
sender_ref.sender_id = sender_id
except TypeError:
pass
else:
self._weak_senders.setdefault(sender_id, sender_ref)
del sender_ref
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
# wire together a cleanup for weakref-able senders
try:
sender_ref = reference(sender, self._cleanup_sender)
sender_ref.sender_id = sender_id
except TypeError:
pass
else:
self._weak_senders.setdefault(sender_id, sender_ref)
del sender_ref
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
# wire together a cleanup for weakref-able senders
try:
sender_ref = reference(sender, self._cleanup_sender)
sender_ref.sender_id = sender_id
except TypeError:
pass
else:
self._weak_senders.setdefault(sender_id, sender_ref)
del sender_ref
`sender=` as a single positional argument and any \*\*kwargs that
were provided to a call to :meth:`send`.
:param sender: Any object or :obj:`ANY`, defaults to ``ANY``.
Restricts notifications delivered to *receiver* to only those
:meth:`send` emissions sent by *sender*. If ``ANY``, the receiver
will always be notified. A *receiver* may be connected to
multiple *sender* values on the same Signal through multiple calls
to :meth:`connect`.
:param weak: If true, the Signal will hold a weakref to *receiver*
and automatically disconnect when *receiver* goes out of scope or
is garbage collected. Defaults to True.
"""
receiver_id = hashable_identity(receiver)
if weak:
receiver_ref = reference(receiver, self._cleanup_receiver)
receiver_ref.receiver_id = receiver_id
else:
receiver_ref = receiver
if sender is ANY:
sender_id = ANY_ID
else:
sender_id = hashable_identity(sender)
self.receivers.setdefault(receiver_id, receiver_ref)
self._by_sender[sender_id].add(receiver_id)
self._by_receiver[receiver_id].add(sender_id)
del receiver_ref
if sender is not ANY and sender_id not in self._weak_senders:
def has_receivers_for(self, sender):
"""True if there is probably a receiver for *sender*.
Performs an optimistic check only. Does not guarantee that all
weakly referenced receivers are still alive. See
:meth:`receivers_for` for a stronger search.
"""
if not self.receivers:
return False
if self._by_sender[ANY_ID]:
return True
if sender is ANY:
return False
return hashable_identity(sender) in self._by_sender