Observers (objetto.observers)

Observer mixin class.

Action Observer Class

class objetto.observers.ActionObserver

Mixin/abstract class for observing an object for actions.

>>> from objetto import Application, Object, attribute
>>> from objetto.observers import ActionObserver

>>> class Person(Object):
...     name = attribute(str, default="Albert")
...
>>> class PersonObserver(ActionObserver):
...
...     def __observe__(self, action, phase):
...         print((action.change.name, phase.value))
...
>>> app = Application()
>>> person = Person(app)
>>> observer = PersonObserver()
>>> token = observer.start_observing(person)
>>> person.name = "Einstein"
('Update Attributes', 'PRE')
('Update Attributes', 'POST')

abstract __observe__(action, phase)

Observe an action (and its execution phase) from an object.

Parameters

action (objetto.objects.Action) – Action.
phase (objetto.constants.PRE or objetto.constants.POST) – Phase.

Raises

NotImplementedError – Abstract method not implemented.

start_observing(obj)

Start observing an object for actions.

Parameters

obj (objetto.objects.Object) – Object.

Returns

Observer token.

Return type

objetto.observers.ActionObserverToken

Raises

TypeError – Invalid ‘obj’ parameter type.
RuntimeError – Can’t start observing while object is initializing.

stop_observing(obj)

Stop observing an object for actions.

Parameters: obj (objetto.objects.Object) – Object.
Raises: TypeError – Invalid ‘obj’ parameter type.

Action Observer Token Class

class objetto.observers.ActionObserverToken(subject, observer)

Allows control over observers’ order/priority.

Inherits from:

objetto.utils.subject_observer.ObserverToken

Note

This class can’t be instantiated directly. A token should be retrieved when an action observer starts observing an object.

>>> from objetto import Application, Object, attribute
>>> from objetto.constants import POST
>>> from objetto.observers import ActionObserver

>>> observed_names = []
...
>>> class Person(Object):
...     name = attribute(str, default="Albert")
...
>>> class PersonObserver(ActionObserver):
...     def __init__(self, this, dependency=None):
...         self.this = this
...         self.dependency = dependency
...
...     def __observe__(self, action, phase):
...         if self.dependency:
...             self.dependency.wait()
...         if phase is POST:
...             observed_names.append(
...                 "{} {}".format(self.this, action.change.new_values["name"])
...             )
...
>>> app = Application()
>>> person = Person(app)

>>> observer_a = PersonObserver("A")
>>> token_a = observer_a.start_observing(person)

>>> observer_b = PersonObserver("B", dependency=token_a)
>>> token_b = observer_b.start_observing(person)

>>> person.name = "Einstein"
>>> observed_names  # note how A will always observe the change before B does
['A Einstein', 'B Einstein']