yet another Observer pattern in python

Observer sniffing a particular attribute from Observable class. This little spy is notified every time watched attribute is changing its value.

Observers are hold in weakref.WeakValueDictionary to avoid crashes in case they would disappear (i.e. being collected and deleted by gc). And here is the source code:

	########################################################################
	# $HeadURL $
	# File: Observer.py
	# Author: Krzysztof.Ciba@NOSPAMgmail.com
	# Date: 2012/09/26 09:39:58
	########################################################################
	""" :mod: Observer
	==============
	.. module: Observer
	:synopsis: Observer pattern
	.. moduleauthor:: Krzysztof.Ciba@NOSPAMgmail.com
	"""
	## for WeakValuesDictionary
	import weakref
	class Observable( type ):
	"""
	.. class:: Observable
	metaclass to create observable pattern
	you can register Observer by calling :registerObserver:
	and remove this little spy by calling :unregisterObserver:
	Observer should implement :notify: method with signature
	notify( attribute, observable, event=None )
	where::
	:param mixed attribute: attribute value
	:param Observable observable: instance you are observing
	:param str event: event send to observer (could be None)
	"""
	def __new__( cls, name, bases, classdict ):
	def observers( self ):
	""" get observers dict """
	if not hasattr( self, "__observers" ):
	setattr( self, "__observers", weakref.WeakValueDictionary() )
	return getattr( self, "__observers" )
	def notifySetAttr( func ):
	""" to be applied exclusively on __setattr__ """
	def wrapper( *args, **kwargs ):
	instance, attr, newVal = args[0], args[1], args[2]
	oldVal = None
	if hasattr( instance, attr ):
	oldVal = getattr( instance, attr )
	ret = func( *args, **kwargs )
	if not oldVal:
	instance.notify( attr, "EVSET" )
	elif oldVal != newVal:
	instance.notify( attr, "EVCHG" )
	return ret
	return wrapper
	def registerObserver( self, observer, watchedAttribute ):
	""" add new :observer: """
	## check for attribute, could raise AttributeError
	getattr( self, watchedAttribute )
	if watchedAttribute not in self.observers():
	self.observers()[watchedAttribute] = observer
	def unregisterObserver( self, watchedAttribute ):
	""" remove :observer: """
	if watchedAttribute in self.observers():
	del self.observers()[watchedAttribute]
	def notify( self, attribute=None, event=None ):
	""" notify observers senfing event :event: """
	if attribute and attribute in self.observers():
	self.observers()[attribute].notify( getattr(self, attribute), self, event )
	else:
	for attribute, observer in self.observers():
	observer.notify( getattr(self, attribute), self, event )
	## add new functions to class dict
	classdict["observers"] = observers
	classdict["registerObserver"] = registerObserver
	classdict["unregisterObserver"] = unregisterObserver
	classdict["notify"] = notify
	aType = type.__new__( cls, name, bases, classdict )
	## decorate setattr to trace down every update of value
	aType.__setattr__ = notifySetAttr( aType.__setattr__ )
	return aType
	class Observer( object ):
	"""
	.. class:: Observer
	dummy observer
	"""
	def notify( self, attribute, observable, event=None ):
	""" callback fruntion from :observable: on :attribute: change """
	raise NotImplementedError("'notify' has to be implemented in the child class")
	## dummy test
	if __name__ == "__main__":
	class TestObservable( object ):
	""" test observable class """
	__metaclass__ = Observable
	def __init__( self ):
	self.x = 1
	def setX( self, val ):
	self.x = val
	class TestObserver( Observer ):
	""" dummy observer """
	def notify( self, attr, caller, event ):
	print "notify called by %s on event %s" % ( caller, event )
	testObserver = TestObserver()
	testObservable = TestObservable()
	testObservable.registerObserver( ao, "x" )
	## triggers notify
	testObservable.x = 2
	## no value change - no notify called
	testObservable.setX(2)
	testObservable.unregisterObserver( "x" )
	## ibserver unregistered - no notify called
	testObservable.x = 5

view raw Observer.py hosted with ❤ by GitHub