Welcome, guest | Sign In | My Account | Store | Cart

I've modified the excellent recipe 576477 to allow for non method functions as well as method functions. This implementation also uses a WeakKeyDictionary instead of a WeakValueDictionary for reasons of code simplification/style.

Python, 100 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
""" A signal/slot implementation

File:    signal.py
Author:  Thiago Marcos P. Santos
Author:  Christopher S. Case
Author:  David H. Bronke
Created: August 28, 2008
Updated: December 12, 2011
License: MIT

"""
from __future__ import print_function
import inspect
from weakref import WeakSet, WeakKeyDictionary


class Signal(object):
    def __init__(self):
        self._functions = WeakSet()
        self._methods = WeakKeyDictionary()

    def __call__(self, *args, **kargs):
        # Call handler functions
        for func in self._functions:
            func(*args, **kargs)

        # Call handler methods
        for obj, funcs in self._methods.items():
            for func in funcs:
                func(obj, *args, **kargs)

    def connect(self, slot):
        if inspect.ismethod(slot):
            if slot.__self__ not in self._methods:
                self._methods[slot.__self__] = set()

            self._methods[slot.__self__].add(slot.__func__)

        else:
            self._functions.add(slot)

    def disconnect(self, slot):
        if inspect.ismethod(slot):
            if slot.__self__ in self._methods:
                self._methods[slot.__self__].remove(slot.__func__)
        else:
            if slot in self._functions:
                self._functions.remove(slot)

    def clear(self):
        self._functions.clear()
        self._methods.clear()


# Sample usage:
if __name__ == '__main__':
    class Model(object):
        def __init__(self, value):
            self.__value = value
            self.changed = Signal()

        def set_value(self, value):
            self.__value = value
            self.changed()  # Emit signal

        def get_value(self):
            return self.__value

    class View(object):
        def __init__(self, model):
            self.model = model
            model.changed.connect(self.model_changed)

        def model_changed(self):
            print("   New value:", self.model.get_value())

    print("Beginning Tests:")
    model = Model(10)
    view1 = View(model)
    view2 = View(model)
    view3 = View(model)

    print("Setting value to 20...")
    model.set_value(20)

    print("Deleting a view, and setting value to 30...")
    del view1
    model.set_value(30)

    print("Clearing all listeners, and setting value to 40...")
    model.changed.clear()
    model.set_value(40)

    print("Testing non-member function...")

    def bar():
        print("   Calling Non Class Function!")

    model.changed.connect(bar)
    model.set_value(50)

I updated this recipe mostly to add support for functions, like lambdas or non class-based handlers. (I find I use them sometimes when writing PyQt code.) My intention is to couple this with a networking library I'm writing, and the functionality would be really useful.

As this uses WeakKeyDictionary and WeakSet, this is a Python 2.7+ implementation only. (I tested in Python 2.7 and Python 3.2) As a note, if you're using this in Python 2.7, you may want to consider changing 'self._methods.items()' to 'self._methods.iteritems()', as it's (slightly) more performant (for a large number of items). In this case, you'd only really want to worry if you have a large number of classes with handler methods connected to the same signal.

This code has been minimally tested, but works great in all of my tests. (I've marked this as requiring Python 3.2, but it may work on older Python 3 implementations.) Credit for the original code goes to Thiago Marcos P. Santos.

2 comments

Ben Strulo 6 years, 6 months ago  # | flag

Thanks for this: it's been helpful.

But a small bug and some other issues came up:

Firstly, the disconnect code is happy if the slot is a function and has not been connected, or if the slot is a method and its __self__ has not been connected. But it raises a KeyError if the slot is a method and its __self__ HAS been connected but its method has not. This doesn't seem good behaviour. For example, this will raise a KeyError in the test code:

model.changed.disconnect(view1.model_changed)
model.changed.disconnect(view1.model_changed)

Then, it is "un-Pythonic" to write code like:

if a in b:
    b.remove(a)

not least because it causes the quite expensive dictionary look-up to be done twice.

Dealing with these two points together suggests the following rewrite of the disconnect method:

def disconnect(self, slot):
    if inspect.ismethod(slot):
        try:
            self._methods[slot.__self__].remove(slot.__func__)
        except KeyError:
            pass
    else:
        try:
            self._functions.remove(slot)
        except KeyError:
            pass

This seems to work fine, but it does have a more marginal issue. In the corner case of objects with methods temporarily connected, the (empty) set of methods remains in the dictionary as long as the object does. This isn't really a memory leak, since the entry will disappear once the object is garbage collected. But it could be a significant overhead: a set for every object that has ever had a connected method. So we might like to get rid of these empty sets when we can, which would give us:

def disconnect(self, slot):
    if inspect.ismethod(slot):
        try:
            s = self._methods[slot.__self__]
            s.remove(slot.__func__)
        except KeyError:
            pass
        else:
            if not s:
                del self._methods[slot.__self__]
    else:
        try:
            self._functions.remove(slot)
        except KeyError:
            pass

This works but it's not as clean as the previous version - for example, it has lost its thread-safety. I can't see any way to avoid that.

Finally, the same un-Pythonic testing is in the connect too. So I would rewrite this to:

def connect(self, slot):
    if inspect.ismethod(slot):
        try:
            self._methods[slot.__self__].add(slot.__func__)
        except KeyError:
            self._methods[slot.__self__] = { slot.__func__ }
    else:
        self._functions.add(slot)

Hope this helps!

Ben Strulo 6 years, 6 months ago  # | flag

Or better still:

def connect(self, slot):
    if inspect.ismethod(slot):
        self._methods.setdefault(slot.__self__,set()).add(slot.__func__)
    else:
        self._functions.add(slot)