bronx.patterns.observer

A personal implementation of the Observer design pattern.

Using the factory get() should provide a convenient way to register to an undetermined number of items hold by ObserverBoard objects.

Example:

# Let's create an observable object
>>> class A(object):
...     def __init__(self, id, status=0):
...         self.id = id
...         self._status = status
...         # Note: the Interface method ``get`` could also be used:
...         self._obs = ObserverBoard(tag='DocTestObsBoard')
...         self._obs.notify_new(self, {'status': self._status})
...     def quit(self):
...         self._obs.notify_del(self, {'status': self._status})
...     def _get_status(self):
...         return self._status
...     def _set_status(self, value):
...         self._status = value
...         self._obs.notify_upd(self, {'status': self._status})
...     status = property(_get_status, _set_status)
...

# Let's create an observer
>>> class MyObserver(Observer):
...     def __init__(self):
...         self.log = list()
...     def newobsitem(self, item, info):
...         self.log.append(('NEW', item.id, info['status']))
...     def updobsitem(self, item, info):
...         self.log.append(('UPD', item.id, info['status']))
...     def delobsitem(self, item, info):
...         self.log.append(('DEL', item.id, info['status']))
...     def __str__(self):
...         return '\n'.join(['{:s} {:6s} status={!s}'.format(*s) for s in self.log])
...

# Create an observer object and register it to the observer board
>>> obs1 = MyObserver()
>>> obs_board = ObserverBoard(tag='DocTestObsBoard')
>>> obs_board.register(obs1)

# Let's create some Observee
>>> a1 = A('First')
>>> a2 = A('Second', status=5)
>>> a2.status = 6
>>> a1.status = 1
>>> a1.quit()
>>> del a1

>>> a2.status = 'z'

# obs1 stops listening
>>> obs_board.unregister(obs1)

# Create a new observer... it will only record new events...
>>> obs2 = MyObserver()
>>> obs_board.register(obs2)

>>> a3 = A('Third')

# List the currently Observed objects
>>> len(obs_board.observed())
2

# List the current Observers
>>> obs_board.observers() # doctest: +ELLIPSIS
[<...MyObserver object at 0x...>]

# What did the observers do ?
>>> print(obs1)
NEW First  status=0
NEW Second status=5
UPD Second status=6
UPD First  status=1
DEL First  status=1
UPD Second status=z
>>> print(obs2)
NEW Third  status=0

Functions

bronx.patterns.observer.get(**kw)

Return an ObserverBoard objects for the specified tag name (a class name for example).

bronx.patterns.observer.items()

Return the items of the ObserverBoard objects collection.

bronx.patterns.observer.keys()

Return actual tags names of the instantiated ObserverBoard objects.

bronx.patterns.observer.values()

Return actual values of the instantiated ObserverBoard objects.

Classes

class bronx.patterns.observer.Observer

Bases: object

Pseudo-Interface class. The three public methods should be implemented by any Observer object.

delobsitem(item, info)

The item has been deleted. Some information is provided through the dict info.

newobsitem(item, info)

A new item has been created. Some information is provided through the dict info.

updobsitem(item, info)

The item has been updated. Some information is provided through the dict info.

class bronx.patterns.observer.ObserverBoard(*args, **kw)

Bases: SecludedObserverBoard, GetByTag

Like a SecludedObserverBoard but using the footprints.util.GetByTag class to provide an easy access to existing boards.

Some class variables may have an impact on GetByTag behaviour:

• _tag_default: Sets the default tag (if the tag attribute is omitted when calling the constructor, the _tag_default string will be used.

• _tag_implicit_new: If set to False, unless new=True is specified when calling the constructor, it won’t be allowed to create new objects (a RuntimeError exception will be thrown).

• _tag_topcls: if set to False, the tags list will be shared with the parent class (and possibly other siblings). The tag_classes() class method allows to retrieve the list of Classes sharing the same list of tags

class bronx.patterns.observer.ParrotObserver

Bases: Observer

Like Observer but boosts the verbosity (useful for tests).

class bronx.patterns.observer.SecludedObserverBoard

Bases: object

A SecludedObserverBoard provides an indirection for the observing pattern.

It holds two lists: one list of objects that are observed and another list of observers, listening to any creation, deletion or update of the observed objects.

notify_del(item, info)

Notify the listening objects that an observed object does not want to be observed anymore.

Note:

It is useless to call notify_del from within an observed object __del__ method. Indeed, When __del__ is called by the garbage collector, the reference count of the observed object is already set to 0 which causes the observed object to disappear from self._items (since self._items is a WeakSet). In turns, since the observed object is not anymore in self._items, this method has no effect.

notify_new(item, info)

Notify the listening objects that a new observed object is born.

notify_upd(item, info)

Notify the listening objects that an observed object has been updated.

observed()

List of observed objects.

observers()

List of observing objects.

register(remote)

Push the remote object to the list of listening objects. A listening object should implement the Observer interface.

unregister(remote)

Remove the remote object from the list of listening objects.