cosapp.core.signal.signal

Module defining the Signal class.

Classes

DummyLock()	Class that implements a no-op instead of a re-entrant lock.
Signal([args, name, threadsafe])	Define a signal by instanciating a Signal object, ie.:

class cosapp.core.signal.signal.DummyLock

Bases: object

Class that implements a no-op instead of a re-entrant lock.

class cosapp.core.signal.signal.Signal(args: List[str] | None = None, name: str | None = None, threadsafe: bool = False)

Bases: object

Define a signal by instanciating a Signal object, ie.:

>>> conf_pre_load = Signal()

Optionaly, you can declare a list of argument names for this signal, ie.:

>>> conf_pre_load = Signal(args=['conf'])

Any callable can be connected to a Signal, it must accept keywords (**kwargs) and be wrapped in a Slot object, ie.:

>>> def yourmodule_conf(conf, **kwargs):
...     conf['yourmodule_option'] = 'foo'
...

Connect your function to the signal using connect():

>>> conf_pre_load.connect(Slot(yourmodule_conf))

Emit the signal to call all connected callbacks using emit():

>>> conf = {}
>>> conf_pre_load.emit(conf=conf)
>>> conf
{'yourmodule_option': 'foo'}

Note that you may disconnect a callback from a signal if it is already connected:

>>> conf_pre_load.is_connected(yourmodule_conf)
True
>>> conf_pre_load.disconnect(yourmodule_conf)
>>> conf_pre_load.is_connected(yourmodule_conf)
False

connect(slot: Slot | Callable) → None

Connect a callback slot to this signal.

disconnect(slot: Slot) → None

Disconnect a slot from a signal if it is connected; else do nothing.

emit(**kwargs) → Any | None

Emit signal, which will execute every connected callback slot, passing keyword arguments.

If a slot returns anything other than None, then emit() will return that value preventing any other slot from being called.

>>> need_something = Signal()
>>> def get_something(**kwargs):
...     return 'got something'
...
>>> def make_something(**kwargs):
...     print('I will not be called')
...
>>> need_something.connect(get_something)
>>> need_something.connect(make_something)
>>> need_something.emit()
'got something'

is_connected(slot: Slot) → bool

Check if a callback slot is connected to this signal.

property slots: List[Slot]

Return a list of slots for this signal.