instrumentable.py

#!/usr/bin/env python
import sys
from timeit import repeat

import blinker


def instrumentable(*event_names):
    def _decorate(_cls):
        _cls._instruments = {n: [] for n in event_names}

        @classmethod
        def add_instrumentation(cls, event_name, fn):
            cls._instruments[event_name].append(fn)

        @classmethod
        def remove_instrumentation(cls, event_name, fn):
            try:
                cls._instruments[event_name].remove(fn)
            except ValueError:
                pass

        def fire_instrumentation(self, event_name, *args, **kwargs):
            if self._instruments[event_name]:
                for fn in self._instruments[event_name]:
                    fn(*args, **kwargs)

        _cls.add_instrumentation = add_instrumentation
        _cls.remove_instrumentation = remove_instrumentation
        _cls.fire_instrumentation = fire_instrumentation

        return _cls
    return _decorate


@instrumentable('event-one', 'event-two')
class Worker(object):
    signal = blinker.Signal()

    def fire_inline(self):
        for fn in self._instruments['event-one']:
            fn()

    def fire_dry(self):
        self.fire_instrumentation('event-one')

    def fire_signal(self):
        self.signal.send()


def run_test(setup, code, label):
    batch = 100000  # less stability higher than this, too fast when lower
    timing = repeat(code, setup, repeat=100, number=batch)
    fastest = (min(timing) / float(batch)) * 1000000000  # nanoseconds
    banner = "%s: %0.2f ns/call" % (label, fastest)
    print(banner)


def test(fn):
    setup = 'from __main__ import Worker; w = Worker()'
    code = 'w.%s()' % fn
    label = "%s" % (fn,)
    run_test(setup, code, label)

try:
    test_fn = sys.argv[1]
except IndexError:
    test_fn = 'fire_inline'

try:
    test(test_fn)
except AttributeError:
    print("unknown test (%s)" % test_fn)

You can optimize a bit more the fire_inline, testing if there are instruments instead of use the for as a conditional, such as:

    def fire_inline(self):
        if self._instruments['event-one']:
            for fn in self._instruments['event-one']:
                fn()

You will get a 185.00 ns per call.