Event system in Python

5 Answers

I've been doing it this way:

class Event(list):
    """Event subscription.

    A list of callable objects. Calling an instance of this will cause a
    call to each item in the list in ascending order by index.

    Example Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> e = Event()
    >>> e()
    >>> e.append(f)
    >>> e(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    >>> del e[0]
    >>> e(2)

    def __call__(self, *args, **kwargs):
        for f in self:
            f(*args, **kwargs)

    def __repr__(self):
        return "Event(%s)" % list.__repr__(self)

However, like with everything else I've seen, there is no auto generated pydoc for this, and no signatures, which really sucks.


What event system for Python do you use? I'm already aware of pydispatcher, but I was wondering what else can be found, or is commonly used?

I'm not interested in event managers that are part of large frameworks, I'd rather use a small bare-bones solution that I can easily extend.

I created an EventManager class (code at the end). The syntax is the following:

#Create an event with no listeners assigned to it
EventManager.addEvent( eventName = [] )

#Create an event with listeners assigned to it
EventManager.addEvent( eventName = [fun1, fun2,...] )

#Create any number event with listeners assigned to them
EventManager.addEvent( eventName1 = [e1fun1, e1fun2,...], eventName2 = [e2fun1, e2fun2,...], ... )

#Add or remove listener to an existing event
EventManager.eventName += extra_fun
EventManager.eventName -= removed_fun

#Delete an event
del EventManager.eventName

#Fire the event

Here is an Example:

def hello(name):
    print "Hello {}".format(name)

def greetings(name):
    print "Greetings {}".format(name)

EventManager.addEvent( salute = [greetings] )
EventManager.salute += hello

print "\nInitial salute"

print "\nNow remove greetings"
EventManager.salute -= greetings


Initial salute
Greetings Oscar
Hello Oscar

Now remove greetings
Hello Oscar

EventManger Code:

class EventManager:

    class Event:
        def __init__(self,functions):
            if type(functions) is not list:
                raise ValueError("functions parameter has to be a list")
            self.functions = functions

        def __iadd__(self,func):
            return self

        def __isub__(self,func):
            return self

        def __call__(self,*args,**kvargs):
            for func in self.functions : func(*args,**kvargs)

    def addEvent(cls,**kvargs):
        addEvent( event1 = [f1,f2,...], event2 = [g1,g2,...], ... )
        creates events using **kvargs to create any number of events. Each event recieves a list of functions,
        where every function in the list recieves the same parameters.


        def hello(): print "Hello ",
        def world(): print "World"

        EventManager.addEvent( salute = [hello] )
        EventManager.salute += world


        Hello World
        for key in kvargs.keys():
            if type(kvargs[key]) is not list:
                raise ValueError("value has to be a list")
                kvargs[key] = cls.Event(kvargs[key])


If you wanted to do more complicated things like merging events or retry you can use the Observable pattern and a mature library that implements that. https://github.com/ReactiveX/RxPY . Observables are very common in Javascript and Java and very convenient to use for some async tasks.

from rx import Observable, Observer

def push_five_strings(observer):

class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.create(push_five_strings)



Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon

Here is a minimal design that should work fine. What you have to do is to simply inherit Observer in a class and afterwards use observe(event_name, callback_fn) to listen for a specific event. Whenever that specific event is fired anywhere in the code (ie. Event('USB connected')), the corresponding callback will fire.

class Observer():
    _observers = []
    def __init__(self):
        self._observed_events = []
    def observe(self, event_name, callback_fn):
        self._observed_events.append({'event_name' : event_name, 'callback_fn' : callback_fn})

class Event():
    def __init__(self, event_name, *callback_args):
        for observer in Observer._observers:
            for observable in observer._observed_events:
                if observable['event_name'] == event_name:


class Room(Observer):
    def __init__(self):
        print("Room is ready.")
        Observer.__init__(self) # DON'T FORGET THIS
    def someone_arrived(self, who):
        print(who + " has arrived!")

# Observe for specific event
room = Room()
room.observe('someone arrived',  room.someone_arrived)

# Fire some events
Event('someone left',    'John')
Event('someone arrived', 'Lenard') # will output "Lenard has arrived!"
Event('someone Farted',  'Lenard')

I use zope.event. It's the most bare bones you can imagine. :-) In fact, here is the complete source code:

subscribers = []

def notify(event):
    for subscriber in subscribers:

Note that you can't send messages between processes, for example. It's not a messaging system, just an event system, nothing more, nothing less.

You may have a look at pymitter (pypi). Its a small single-file (~250 loc) approach "providing namespaces, wildcards and TTL".

Here's a basic example:

from pymitter import EventEmitter

ee = EventEmitter()

# decorator usage
def handler1(arg):
   print "handler1 called with", arg

# callback usage
def handler2(arg):
    print "handler2 called with", arg
ee.on("myotherevent", handler2)

# emit
ee.emit("myevent", "foo")
# -> "handler1 called with foo"

ee.emit("myotherevent", "bar")
# -> "handler2 called with bar"