Which Python packages offer a stand-alone event system?

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)
    f(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    f(10)
    g(10)
    >>> del e[0]
    >>> e(2)
    g(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.

We use an EventHook as suggested from Michael Foord in his Event Pattern:

Just add EventHooks to your classes with:

class MyBroadcaster()
    def __init__():
        self.onChange = EventHook()

theBroadcaster = MyBroadcaster()

# add a listener to the event
theBroadcaster.onChange += myFunction

# remove listener from the event
theBroadcaster.onChange -= myFunction

# fire event
theBroadcaster.onChange.fire()

We add the functionality to remove all listener from an object to Michaels class and ended up with this:

class EventHook(object):

    def __init__(self):
        self.__handlers = []

    def __iadd__(self, handler):
        self.__handlers.append(handler)
        return self

    def __isub__(self, handler):
        self.__handlers.remove(handler)
        return self

    def fire(self, *args, **keywargs):
        for handler in self.__handlers:
            handler(*args, **keywargs)

    def clearObjectHandlers(self, inObject):
        for theHandler in self.__handlers:
            if theHandler.im_self == inObject:
                self -= theHandler

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:
        subscriber(event)

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.

I found this small script on Valued Lessons. It seems to have just the right simplicity/power ratio I'm after. Peter Thatcher is the author of following code (no licensing is mentioned).

class Event:
    def __init__(self):
        self.handlers = set()

    def handle(self, handler):
        self.handlers.add(handler)
        return self

    def unhandle(self, handler):
        try:
            self.handlers.remove(handler)
        except:
            raise ValueError("Handler is not handling this event, so cannot unhandle it.")
        return self

    def fire(self, *args, **kargs):
        for handler in self.handlers:
            handler(*args, **kargs)

    def getHandlerCount(self):
        return len(self.handlers)

    __iadd__ = handle
    __isub__ = unhandle
    __call__ = fire
    __len__  = getHandlerCount

class MockFileWatcher:
    def __init__(self):
        self.fileChanged = Event()

    def watchFiles(self):
        source_path = "foo"
        self.fileChanged(source_path)

def log_file_change(source_path):
    print "%r changed." % (source_path,)

def log_file_change2(source_path):
    print "%r changed!" % (source_path,)

watcher              = MockFileWatcher()
watcher.fileChanged += log_file_change2
watcher.fileChanged += log_file_change
watcher.fileChanged -= log_file_change2
watcher.watchFiles()

If I do code in pyQt I use QT sockets/signals paradigm, same is for django

If I'm doing async I/O I use native select module

If I'm usign a SAX python parser I'm using event API provided by SAX. So it looks like I'm victim of underlying API :-)

Maybe you should ask yourself what do you expect from event framework/module. My personal preference is to use Socket/Signal paradigm from QT. more info about that can be found here

Here's another module for consideration. It seems a viable choice for more demanding applications.

Py-notify is a Python package providing tools for implementing Observer programming pattern. These tools include signals, conditions and variables.

Signals are lists of handlers that are called when signal is emitted. Conditions are basically boolean variables coupled with a signal that is emitted when condition state changes. They can be combined using standard logical operators (not, and, etc.) into compound conditions. Variables, unlike conditions, can hold any Python object, not just booleans, but they cannot be combined.

If you need an eventbus that works across process or network boundaries you can try PyMQ. It currently supports pub/sub, message queues and synchronous RPC. The default version works on top of a Redis backend, so you need a running Redis server. There is also an in-memory backend for testing. You can also write your own backend.

import pymq

# common code
class MyEvent:
    pass

# subscribe code
@pymq.subscriber
def on_event(event: MyEvent):
    print('event received')

# publisher code
pymq.publish(MyEvent())

# you can also customize channels
pymq.subscribe(on_event, channel='my_channel')
pymq.publish(MyEvent(), channel='my_channel')

To initialize the system:

from pymq.provider.redis import RedisConfig

# starts a new thread with a Redis event loop
pymq.init(RedisConfig())

# main application control loop

pymq.shutdown()

Disclaimer: I am the author of this library

Another handy package is events. It encapsulates the core to event subscription and event firing and feels like a "natural" part of the language. It seems similar to the C# language, which provides a handy way to declare, subscribe to and fire events. Technically, an event is a "slot" where callback functions (event handlers) can be attached to - a process referred to as subscribing to an event.

# Define a callback function
def something_changed(reason):
    print "something changed because %s" % reason

# Use events module to create an event and register one or more callback functions
from events import Events
events = Events()
events.on_change += something_changed

When the event is fired, all attached event handlers are invoked in sequence. To fire the event, perform a call on the slot:

events.on_change('it had to happen')

This will output:

'something changed because it had to happen'

More documentation can be found in the github repo or the documentation.

You can try buslane module.

This library makes implementation of message-based system easier. It supports commands (single handler) and events (0 or multiple handlers) approach. Buslane uses Python type annotations to properly register handler.

Simple example:

from dataclasses import dataclass

from buslane.commands import Command, CommandHandler, CommandBus


@dataclass(frozen=True)
class RegisterUserCommand(Command):
    email: str
    password: str


class RegisterUserCommandHandler(CommandHandler[RegisterUserCommand]):

    def handle(self, command: RegisterUserCommand) -> None:
        assert command == RegisterUserCommand(
            email='john@lennon.com',
            password='secret',
        )


command_bus = CommandBus()
command_bus.register(handler=RegisterUserCommandHandler())
command_bus.execute(command=RegisterUserCommand(
    email='john@lennon.com',
    password='secret',
))

To install buslane, simply use pip:

$ pip install buslane

Some time ago I've wrote library that might be useful for you. It allows you to have local and global listeners, multiple different ways of registering them, execution priority and so on.

from pyeventdispatcher import register

register("foo.bar", lambda event: print("second"))
register("foo.bar", lambda event: print("first "), -100)

dispatch(Event("foo.bar", {"id": 1}))
# first second

Have a look pyeventdispatcher