DOCSEARCHbyAlgolia
Submit Docs

blinker

repository·main·Indexed 24 days ago

https://github.com/pallets-eco/blinker

A fast Python dispatching system implementing a signal/receiver pattern for decoupled component communication via events. It supports named and anonymous signals, specific sender subscriptions, asynchronous receivers via send_async(), and context managers for muting or temporary connections. Key components include Signal, NamedSignal, and Namespace.

Tokens
3.1K
Snippets
9
Records
22
Agent score
85%

What's inside blinker

  2. Create anonymous signals

    main

    If you do not need a global registry, you can create unique, anonymous signals by instantiating the Signal class directly. These signals are not shared via a name registry; every instance is unique.

    from blinker import Signal

class AltProcessor:
    on_ready = Signal()
    on_complete = Signal()

    def go(self):
        self.on_ready.send(self)

  3. Call receivers in order of registration

    main

    By default, receivers are stored in an unordered set. If you need receivers to be called in the exact order they were registered, you must change the Signal.set_class attribute to an ordered set implementation (e.g., from the ordered-set package).

    from blinker import Signal
from ordered_set import OrderedSet

# All future Signal instances will use OrderedSet for receivers
Signal.set_class = OrderedSet

  4. How to use Blinker signals and receivers

    main

    You can create a signal using signal(name) and then connect receiver functions to it using the .connect() method.

    Receivers can be subscribed to:

    1. All signals of a specific name: By calling .connect(receiver) without a sender argument.
    2. Specific senders: By calling .connect(receiver, sender=some_object). In this case, the receiver will only be triggered if the signal is sent specifically by some_object.

    To trigger the signal and notify all connected receivers, use the .send(*args, **kwargs) method.

    from blinker import signal

# Create a signal
started = signal('round-started')

# Define a receiver
def each(round):
    print(f"Round {round}")

# Connect the receiver to the signal (receives all 'round-started' signals)
started.connect(each)

# Define another receiver for a specific sender
def round_two(round):
    print("This is round two.")

# Connect with a specific sender constraint
started.connect(round_two, sender=2)

# Trigger the signal
for round in range(1, 4):
    started.send(round)

  5. Use connect_via() for advanced decorator connections

    main

    If you need to use a decorator but also need to specify a sender or other arguments, use Signal.connect_via(). This allows you to chain multiple connect_via calls to register the same function for different senders.

    from blinker import signal

dice_roll = signal('dice_roll')

@dice_roll.connect_via(1)
@dice_roll.connect_via(3)
@dice_roll.connect_via(5)
def odd_subscriber(sender):
    print(f"Observed dice roll {sender!r}.")

dice_roll.send(3)

  6. Subscribe to signals using connect()

    main

    To react to a signal, use the Signal.connect() method to register a receiver function. The receiver function is always passed the sender object (the object that triggered the signal) as its first argument.

    from blinker import signal

def subscriber(sender):
    print(f"Got a signal sent by {sender!r}")

ready = signal('ready')
ready.connect(subscriber)

  7. Emit signals with send()

    main

    Use Signal.send() to notify all connected receivers. You should pass the object responsible for the event as the first argument to send() so receivers know who the sender is. If no receivers are connected, send() performs an optimized no-op.

    from blinker import signal

class Processor:
    def go(self):
        ready = signal('ready')
        ready.send(self)

processor = Processor()
processor.go()

  9. Mute signals using a context manager

    main

    To temporarily disable a signal (useful for unit testing), use the Signal.muted() method as a context manager. While inside the with block, signal emissions will not trigger any receivers.

    from blinker import signal

sig = signal('send-data')
with sig.muted():
    # Signals emitted here will not be received
    sig.send('sender')

  10. Create named signals with signal()

    main
    Named signals allow different parts of your application to communicate without direct imports or code sharing. You create a named signal using signal('name'). Every subsequent call to signal('name') with the same name returns the exact same signal object, acting as a global registry.

  11. Send and receive data through signals

    main

    You can pass arbitrary keyword arguments to Signal.send(). These are passed directly to the connected receiver functions. Signal.send() returns a list of (receiver_function, return_value) pairs for all connected receivers.

    from blinker import signal

send_data = signal('send-data')

@send_data.connect
def receive_data(sender, **kw):
    print(f"Caught signal from {sender!r}, data {kw!r}")
    return 'received!'

# 'abc=123' is passed to receive_data as a keyword argument
result = send_data.send('anonymous', abc=123)
# result is: [(<function receive_data at ...>, 'received!')]

  12. Subscribe to specific senders

    main

    By default, a connected receiver is invoked for any sender of that signal. To restrict a subscription so a receiver only triggers when a specific object emits the signal, use the sender keyword argument in connect().

    from blinker import signal

def b_subscriber(sender):
    print("Caught signal from processor_b.")

processor_b = object()
ready = signal('ready')

# This receiver only triggers if processor_b is the sender
ready.connect(b_subscriber, sender=processor_b)