Connect Python objects to blinker signals

I started using blinker for handroll. Blinker is a signal generation library for broadcasting events. The library lets signalers send messages to connected receiver functions. I will explain how I convinced Blinker to talk to objects instead of pure Python functions.

The example code is going to handle a "frobnicated" signal. Remember, the signal itself is not very important.

import blinker

frobnicated = blinker.signal('frobnicated')

frobnicated is a named signal. In a real project, you might put all your signals in a single module. Pelican does this nicely. Grouping all your signals in one place gives signal consumers a clear view of what is available.

class Receiver(object):

    def __init__(self):
        def handle_frobnicated(sender, **kwargs):
            self.on_frobnicated(sender, **kwargs)
        self.handle_frobnicated = handle_frobnicated
        frobnicated.connect(handle_frobnicated)

    def on_frobnicated(self, sender, **kwargs):
        print(sender, kwargs['message'])

The __init__ method is where all the magic happens. The first thing to notice is the use of an inner function, handle_frobnicated. The inner function uses the method signature that the signal will invoke, and delegates to Receiver.on_frobnicated. Why? This is necessary because Blinker can't pass self to receiver functions. handle_frobnicated acts as a closure on self which lets the signal call the instance method.

        self.handle_frobnicated = handle_frobnicated

That seems like a strange line, doesn't it? Blinker does some funny stuff with references. Without storing the inner function, Blinker will delete a weak function reference and the inner function will no longer be among the signal's receivers. I stared at the Blinker source code for a long time to figure that mystery out.

The last line in __init__ connects the signal to the inner function. The receiver is ready to handle frobnicated events.

if __name__ == '__main__':
    receiver = Receiver()
    for i in range(10):
        frobnicated.send('Sender %s' % i, message='hello')

The code to fire the signal is fairly boring. Notice that frobnicated.send has no need for receiver. The publisher is disconnected from subscriber at this stage. The final result looks like:

$ python blink_object.py
Sender 0 hello
Sender 1 hello
Sender 2 hello
Sender 3 hello
Sender 4 hello
Sender 5 hello
Sender 6 hello
Sender 7 hello
Sender 8 hello
Sender 9 hello

By connecting a signal to an object, you get all the benefits that come along with classes. Rather than making a monsterous function, you could use various instance methods within the handler. This flexibility is a boon for unit testing. The gain has similar advantages to using class based views in Django rather than function views.

Here's the full example.

import blinker

frobnicated = blinker.signal('frobnicated')


class Receiver(object):

    def __init__(self):
        def handle_frobnicated(sender, **kwargs):
            self.on_frobnicated(sender, **kwargs)
        self.handle_frobnicated = handle_frobnicated
        frobnicated.connect(handle_frobnicated)

    def on_frobnicated(self, sender, **kwargs):
        print(sender, kwargs['message'])


if __name__ == '__main__':
    receiver = Receiver()
    for i in range(10):
        frobnicated.send('Sender %s' % i, message='hello')

This article first appeared on mattlayman.com.