Python Observer Pattern: Examples, Tips? [closed]

However it does lack flexibility.

Well... actually, this looks like a good design to me if an asynchronous API is what you want. It usually is. Maybe all you need is to switch from stderr to Python's logging module, which has a sort of publish/subscribe model of its own, what with Logger.addHandler() and so on.

If you do want to support observers, my advice is to keep it simple. You really only need a few lines of code.

class Event(object):    passclass Observable(object):    def __init__(self):        self.callbacks = []    def subscribe(self, callback):        self.callbacks.append(callback)    def fire(self, **attrs):        e = Event()        e.source = self        for k, v in attrs.iteritems():            setattr(e, k, v)        for fn in self.callbacks:            fn(e)

Your Job class can subclass Observable. When something of interest happens, call self.fire(type="progress", percent=50) or the like.

I think people in the other answers overdo it. You can easily achieve events in Python with less than 15 lines of code.

You simple have two classes: Event and Observer. Any class that wants to listen for an event, needs to inherit Observer and set to listen (observe) for a specific event. When an Event is instantiated and fired, all observers listening to that event will run the specified callback functions.

class Observer():    _observers = []    def __init__(self):        self._observers.append(self)        self._observables = {}    def observe(self, event_name, callback):        self._observables[event_name] = callbackclass Event():    def __init__(self, name, data, autofire = True):        self.name = name        self.data = data        if autofire:            self.fire()    def fire(self):        for observer in Observer._observers:            if self.name in observer._observables:                observer._observables[self.name](self.data)

Example:

class Room(Observer):    def __init__(self):        print("Room is ready.")        Observer.__init__(self) # Observer's init needs to be called    def someone_arrived(self, who):        print(who + " has arrived!")room = Room()room.observe('someone arrived',  room.someone_arrived)Event('someone arrived', 'Lenard')

Output:

Room is ready.Lenard has arrived!

A few more approaches...

Example: the logging module

Maybe all you need is to switch from stderr to Python's logging module, which has a powerful publish/subscribe model.

It's easy to get started producing log records.

# producerimport logginglog = logging.getLogger("myjobs")  # that's all the setup you needclass MyJob(object):    def run(self):        log.info("starting job")        n = 10        for i in range(n):            log.info("%.1f%% done" % (100.0 * i / n))        log.info("work complete")

On the consumer side there's a bit more work. Unfortunately configuring logger output takes, like, 7 whole lines of code to do. ;)

# consumerimport myjobs, sys, loggingif user_wants_log_output:    ch = logging.StreamHandler(sys.stderr)    ch.setLevel(logging.INFO)    formatter = logging.Formatter(        "%(asctime)s - %(name)s - %(levelname)s - %(message)s")    ch.setFormatter(formatter)    myjobs.log.addHandler(ch)    myjobs.log.setLevel(logging.INFO)myjobs.MyJob().run()

On the other hand there's an amazing amount of stuff in the logging package. If you ever need to send log data to a rotating set of files, an email address, and the Windows Event Log, you're covered.

Example: simplest possible observer

But you don't need to use any library at all. An extremely simple way to support observers is to call a method that does nothing.

# producerclass MyJob(object):    def on_progress(self, pct):        """Called when progress is made. pct is the percent complete.        By default this does nothing. The user may override this method        or even just assign to it."""        pass    def run(self):        n = 10        for i in range(n):            self.on_progress(100.0 * i / n)        self.on_progress(100.0)# consumerimport sys, myjobsjob = myjobs.MyJob()job.on_progress = lambda pct: sys.stdout.write("%.1f%% done\n" % pct)job.run()

Sometimes instead of writing a lambda, you can just say job.on_progress = progressBar.update, which is nice.

This is about as simple as it gets. One drawback is that it doesn't naturally support multiple listeners subscribing to the same events.

Example: C#-like events

With a bit of support code, you can get C#-like events in Python. Here's the code:

# glue codeclass event(object):    def __init__(self, func):        self.__doc__ = func.__doc__        self._key = ' ' + func.__name__    def __get__(self, obj, cls):        try:            return obj.__dict__[self._key]        except KeyError, exc:            be = obj.__dict__[self._key] = boundevent()            return beclass boundevent(object):    def __init__(self):        self._fns = []    def __iadd__(self, fn):        self._fns.append(fn)        return self    def __isub__(self, fn):        self._fns.remove(fn)        return self    def __call__(self, *args, **kwargs):        for f in self._fns[:]:            f(*args, **kwargs)

The producer declares the event using a decorator:

# producerclass MyJob(object):    @event    def progress(pct):        """Called when progress is made. pct is the percent complete."""    def run(self):        n = 10        for i in range(n+1):            self.progress(100.0 * i / n)#consumerimport sys, myjobsjob = myjobs.MyJob()job.progress += lambda pct: sys.stdout.write("%.1f%% done\n" % pct)job.run()

This works exactly like the "simple observer" code above, but you can add as many listeners as you like using +=. (Unlike C#, there are no event handler types, you don't have to new EventHandler(foo.bar) when subscribing to an event, and you don't have to check for null before firing the event. Like C#, events do not squelch exceptions.)

How to choose

If logging does everything you need, use that. Otherwise do the simplest thing that works for you. The key thing to note is that you don't need to take on a big external dependency.