Skip to main content

Signal

Base class for all signals

Attributes

AttributeTypeDescription
receiverslist = []Internal list of registered receiver metadata, including lookup keys, weak references to receivers and senders, and asynchronicity flags.
lockthreading.Lock = threading.Lock()A threading lock used to ensure thread-safe access when modifying receivers or clearing the sender cache.
use_cachingboolean = FalseBoolean flag that determines whether to cache the resolved receivers for each distinct sender to improve performance.
sender_receivers_cacheWeakKeyDictionary or dict = {}A mapping of senders to their resolved receivers, implemented as a WeakKeyDictionary if caching is enabled.
_dead_receiversboolean = FalseInternal flag indicating that the receivers list contains dead weak references that need to be cleaned up.

Constructor

Signature

def Signal(
use_caching: boolean = False
) - > None

Parameters

NameTypeDescription
use_cachingboolean = FalseIf set to True, the signal will cache the receivers for each distinct sender to improve performance on subsequent sends.

Methods


connect()

@classmethod
def connect(
receiver: callable,
sender: object,
weak: boolean,
dispatch_uid: hashable
) - > null

Connect receiver to sender for signal.

Parameters

NameTypeDescription
receivercallableA function or instance method that will be called when the signal is emitted; must accept keyword arguments.
senderobjectThe specific object from which the receiver should respond to events, or None to listen to all senders.
weakbooleanWhether to use weak references to the receiver to allow it to be garbage collected automatically.
dispatch_uidhashableA unique identifier to prevent the same receiver from being subscribed multiple times.

Returns

TypeDescription
null

disconnect()

@classmethod
def disconnect(
receiver: callable,
sender: object,
dispatch_uid: hashable
) - > boolean

Disconnect receiver from sender for signal.

Parameters

NameTypeDescription
receivercallableThe registered receiver function or method to remove from the signal's dispatch list.
senderobjectThe registered sender from which the receiver should be disconnected.
dispatch_uidhashableThe unique identifier of the specific receiver instance to disconnect.

Returns

TypeDescription
booleanTrue if a receiver was successfully disconnected, False otherwise.

has_listeners()

@classmethod
def has_listeners(
sender: object
) - > boolean

Checks if there are any active receivers currently connected to the signal for a given sender.

Parameters

NameTypeDescription
senderobjectThe object emitting the signal to check for associated listeners.

Returns

TypeDescription
booleanTrue if at least one synchronous or asynchronous listener is registered for the sender.

send()

@classmethod
def send(
sender: object,
**named: dict
) - > list

Send signal from sender to all connected receivers.

Parameters

NameTypeDescription
senderobjectThe object initiating the signal dispatch.
**nameddictArbitrary keyword arguments passed directly to every connected receiver.

Returns

TypeDescription
listA list of tuple pairs containing the receiver and its corresponding response value.

asend()

@classmethod
def asend(
sender: object,
**named: dict
) - > list

Send signal from sender to all connected receivers in async mode.

Parameters

NameTypeDescription
senderobjectThe object initiating the signal dispatch.
**nameddictArbitrary keyword arguments passed directly to every connected receiver.

Returns

TypeDescription
listA list of tuple pairs containing the receiver and its corresponding response value, resolved asynchronously.

send_robust()

@classmethod
def send_robust(
sender: object,
**named: dict
) - > list

Send signal from sender to all connected receivers catching errors.

Parameters

NameTypeDescription
senderobjectThe object initiating the signal dispatch.
**nameddictArbitrary keyword arguments passed directly to every connected receiver.

Returns

TypeDescription
listA list of tuple pairs containing the receiver and either its response or the Exception instance if it failed.

asend_robust()

@classmethod
def asend_robust(
sender: object,
**named: dict
) - > list

Send signal from sender to all connected receivers catching errors.

Parameters

NameTypeDescription
senderobjectThe object initiating the signal dispatch.
**nameddictArbitrary keyword arguments passed directly to every connected receiver.

Returns

TypeDescription
listA list of tuple pairs containing the receiver and either its response or the Exception instance if it failed, resolved asynchronously.