onoff.py - Add on(), off(), and trigger() event handling to any Python class¶

Module author: Dan McDougall <daniel.mcdougall@liftoffsoftware.com>

onoff.py - A really handy way to add on(), off(), and trigger() styled events to any Python class. A quick example:

>>> from onoff import OnOffMixin
>>> class Foo(OnOffMixin):
...     def __init__(self):
...         self.on("hello", self.hello)
...     def hello(self, *args):
...         print("Hello: %s" % args)
...     def test(self, *args):
...         self.trigger("hello", *args)
...
>>> f = Foo()
>>> f.test("Triggered events rock!")
Hello: Triggered events rock!

class onoff.OnOffMixin[source]¶

A mixin to add on(), off(), and trigger() event handling methods to any class.

For an example, let's pretend we've got a basic WebSocket server that can perform a number of functions based on the incoming message:

class ActionWebSocket(WebSocketHandler):
    def open(self):
        print("WebSocket opened")

    def on_message(self, message):
        if message == 'hello':
            self.hello()
        elif message == 'ping':
            self.pong()

    def on_close(self):
        print("WebSocket closed")

    def pong(self, timestamp):
        self.write_message('pong')

    def hello(self):
        self.write_message('Hey there!')

This works OK for the most simple of stuff. We could use string parsing of various sorts (startswith(), json, etc) to differentiate messages from each other but our conditionals will quickly grow into a giant mess.

There's a better way, using the OnOffMixin:

class ActionWebSocket(WebSocketHandler, OnOffMixin):
    "Calls an appropriate 'action' based on the incoming message."
    def __init__(self, *args, **kwargs):
        super(ActionWebSocket, self).__init__(*args, **kwargs)
        self.on("ping", self.pong)      # Register a "ping" event
        self.on("hello", self.heythere) # Register a "hello" event

    def on_message(self, message):
        # Assume we're sent a json-encoded dict:
        message_obj = json.loads(message)
        # message_obj would be something like '{"eventname": <args>}'
        for key, value in message_obj.items(): # If more than one
            self.trigger(key, value)

    def pong(self, timestamp):
        "Responds to meesages like ``{"ping": <timestamp>}``"
        self.write_message('{"pong": %s}' % timestamp)

    def heythere(self, *args):
        print("heythere() got args: %s" % args)
        for arg in args:
            self.write_message('Hey there, %s!' % arg)

In the above example we used the OnOffMixin to add on(), off(), and trigger() methods to our ActionWebSocket class.

off(events, callback)[source]¶

Removes the given callback from the given events (string or list of strings).

on(events, callback, times=None)[source]¶

Registers the given callback with the given events (string or list of strings) that will get called whenever the given event is triggered (using self.trigger()).

If times is given the callback will only be fired that many times before it is automatically removed from self._on_off_events.

once(events, callback)[source]¶

A shortcut for self.on(events, callback, 1)

trigger(events, *args, **kwargs)[source]¶

Fires the given events (string or list of strings). All callbacks associated with these events will be called and if their respective objects have a times value set it will be used to determine when to remove the associated callback from the event.

If given, callbacks associated with the given events will be called with args and kwargs.