Skip to main content

Return to Answer

Post Timeline

add a paragraph on how to choose
Source Link
Jason Orendorff
Jason Orendorff
  • 45.5k
  • 6
  • 69
  • 106
# producer class 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) #consumer import sys, myjobs job = 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.

# producer class 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) #consumer import sys job = 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.)

# producer class 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) #consumer import sys, myjobs job = 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.

Source Link
Jason Orendorff
Jason Orendorff
  • 45.5k
  • 6
  • 69
  • 106

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.

# producer import logging log = logging.getLogger("myjobs") # that's all the setup you need class 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. ;)

# consumer import myjobs, sys, logging if 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.

# producer class 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) # consumer import sys, myjobs job = 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 code class 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 be class 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:

# producer class 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) #consumer import sys job = 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.)