Signaller - реализация механизма "сигналы-слоты" с поддержкой asyncio

from signaller import Signal

logging.basicConfig(level=logging.DEBUG)

def slot(arg):
    print('slot:', arg)

# Creating signals (you can set signal name, but it is not required,
# signals can be anonymous):
sig_test = Signal('sig_test')

# Connecting signals to slots (uses weak references by default,
# but you can force strong references by specifying weak=False):
sig_test.connect(slot)
sig_test.connect(lambda arg: print('slot_lambda:', arg), weak=False)

# You can also use decorators for connecting signals to slots:
@sig_test.connect
def slot2(arg):
    print('slot2:', arg)

# And weak=False can be specified when using decorators too:
@sig_test.connect(weak=False)
def slot3(arg):
    print('slot3:', arg)

# Slots are automatically disconnected from signals
# when using weak references:
del slot2

# Or you can disconnect slots manually:
sig_test.disconnect(slot3)

# Emitting signals (you can send positional and keyword
# arguments to connected slots):
sig_test.emit('Hello world!')