Skip to content

Instantly share code, notes, and snippets.

@RemyPorter
Created November 26, 2014 16:51
Show Gist options
  • Save RemyPorter/784cab94a29508bcec5c to your computer and use it in GitHub Desktop.
Save RemyPorter/784cab94a29508bcec5c to your computer and use it in GitHub Desktop.
An example of implementing the observer pattern using decorators. I'm sure there's a secret crime in implementing GOF patterns this way.
"""The observer pattern, implemented as decorators,
to simplify event-binding. Usage:
EventSender("foo")
class FooSender:
def some_method(self):
self.send_foo_event(details)
EventReceiver("foo","foo_handler")
class FooReceiver:
def foo_handler(self, sender, details):
sender = FooSender()
receiver = FooReceiver()
receiver.register_foo(sender)
sender.some_method() #sends the event notification to all subscribers
"""
_observers_prop = "__{0}_observers"
_add_prop = "add_{0}_observer"
_rem_prop = "remove_{0}_observer"
_sen_prop = "send_{0}_event"
_get_prop = "get_{0}_observers"
_handle_prop = "{0}_event"
_sub_prop = "register_{0}"
_unsub_prop = "unregister_{0}"
def EventSender(event_name):
"""This decorator will add a suite of methods to an object
to help it send notifications to observer instances. It
is, essentially, an observable interface.
Given an event name, "foo", it will add the following methods
to the class it decorates:
add_foo_observer, remove_foo_observer, get_foo_observers,
and most important, send_foo_event, which blasts a notification
out to the observers."""
def wrapper(original_class):
init = original_class.__init__
observers_prop = _observers_prop.format(event_name)
add_prop = _add_prop.format(event_name)
rem_prop = _rem_prop.format(event_name)
sen_prop = _sen_prop.format(event_name)
get_prop = _get_prop.format(event_name)
def __init__(self, *args, **kwargs):
init(self, *args, **kwargs)
setattr(self, observers_prop, set())
original_class.__init__ = __init__
def add(self, observer):
"""Add an observer for the event."""
obs_list = getattr(self, observers_prop)
obs_list.add(observer)
def remove(self, observer):
"""Remove an observer from the set of observers."""
obs_list = getattr(self, observers_prop)
obs_list.remove(observer)
def notify(self, event_details):
"""Notify the observers that an event has occurred."""
obs_list = getattr(self, observers_prop)
for o in obs_list:
notify_method = getattr(o,
_handle_prop.format(event_name))
notify_method(self, event_details)
def get_observers(self):
"""Get a copy of the set of observers."""
return set(getattr(self, observers_prop))
setattr(original_class, add_prop, add)
setattr(original_class, rem_prop, remove)
setattr(original_class, sen_prop, notify)
setattr(original_class, get_prop, get_observers)
return original_class
return wrapper
def EventReceiver(event_name, behavior_name):
"""This decorator simplifies handling events generated by the
EventSender. This binds a behavior to the event-handling name"""
def wrapper(original_class):
handler = getattr(original_class, behavior_name)
setattr(original_class, _handle_prop.format(event_name),
handler)
def register(self, sender):
meth = getattr(sender, _add_prop.format(event_name))
meth(self)
def unregister(self, sender):
meth = getattr(sender, _rem_prop.format(event_name))
meth(self)
setattr(original_class, _sub_prop.format(event_name), register)
setattr(original_class, _unsub_prop.format(event_name),
unregister)
return original_class
return wrapper
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment