Welcome, guest | Sign In | My Account | Store | Cart

This recipe uses descriptors of new style classes to implement the observer pattern. This implementation supports exactly one observer (no one-many relation).

This recipe builds on recipe http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/306864. It assumes that the text of the recipe is stored as "list_dict_observer.py".

Python, 179 lines
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
"""
Implement an observer pattern for attribute access.

Assumes that each attribute is used (exclusivly) for either scalar
values, lists or maps.

It does not work if you store a scalar in an attribute, and then later
a map or list.

A scalar attribute is an attribute whose values do not have internal
structure.  Scalars are integers, floats, (short) strings, and
references to instances and classes.

Provides some support to observe lists and dictionaries which are
assigned to attributes.

Usage pattern:
    class c(object):
        ...some definitions...
    
    class observer_class(object):
        ...some definitions...
        
    observer = observer_class()
        
    c.x = scalar_observer("x", observer)
    i = c()
    i.x = 13
    # The assignment above will trigger a method call to observer.
    
    For lists, you can use:
        
    c.l = list_observer_in_instance("l", observer)
    i = c()
    i.l = l = [1, [1, 2], 3]
    i.l.append(4)
    # The append above will trigger a method call to observer.
    # Two caveats here:
    #     1. i.l now points to a different list than l.
    #        (you should probably avoid such aliasing).
    #        If you want to continue to use l, you should
    #        probably add:
    #        l = i.l
    #     2. The inner list [1, 2] is *not* monitored.
    #        (if desired, invoke list_observer([1, 2], observer) instead
    #         of just [1, 2]).
    
The attribute is replaced by a descriptor in the class. The attributes
in the instances have too leading underscores (__).
"""

from list_dict_observer import list_observer, dict_observer, printargs
    
class scalar_observer(object):
    """
    Observes a scalar attribute.
    
    Can be used for integers, longs, floats, (short) strings, and
    references to instances or classes.
    
    Lists and dictionaries have specialized classes.
    
    All assignments to the attributes are intercepted by
    descriptors. The values themselves are stored in an attribute with
    a name __<attributename>.
    """
    def __init__ (self, external_attributename, observer):
        self.external_attributename = external_attributename 
        self.private_attributename = '__'+external_attributename 
        self.observer = observer 
    
    def __set__ (self, instance, value):
        private_attributename = self.private_attributename 
        external_attributename = self.external_attributename 
        try:
            oldvalue = getattr(instance, private_attributename)
        except AttributeError:
            setattr(instance, private_attributename, value)
            self.observer.scalar_set(instance, private_attributename, external_attributename)
        else:
            if oldvalue!=value:
                setattr(instance, private_attributename, value)
                self.observer.scalar_modify(instance, private_attributename, external_attributename, oldvalue)
    def __get__ (self, instance, owner):
        return getattr(instance, self.private_attributename)
    
class list_observer_in_instance(object):
    """
    Observes instance attributes which contain a list as a value.
    
    Assignments to these attributes, which must be lists, are replaced by instances of 'list_observer'.
    
    Note that you are not notified by changes to inner lists or maps.
    
    You should also be aware of aliasing.
    """
    
    def __init__ (self, external_attributename, observer):
        self.external_attributename = external_attributename 
        self.internal_attributename = '__'+external_attributename 
        self.observer = observer 
    
    def __set__ (self, instance, value):
        """Intercept assignments to the external attribute"""
        assert isinstance(value, type([]))
        if isinstance(value, list_observer):
            newvalue = value 
            # if the value is already a list observer, assume that this value
            # already has an observer. Do new create a new list in this case.
        else:
            newvalue = list_observer(value, self.observer)
        internal_attributename = self.internal_attributename 
        try:
            oldvalue = getattr(instance, internal_attributename)
        except AttributeError:
            self.observer.list_assignment_new(instance, internal_attributename)
        else:
            self.observer.list_assignment_replace(instance, internal_attributename, oldvalue)
        setattr(instance, self.internal_attributename, newvalue)
    
    def __get__ (self,instance,owner):
        try:
            return instance.__dict__[self.internal_attributename]
        except KeyError:
            return instance.__dict__[self.external_attributename]
    
class dict_observer_in_instance(object):
    """
    observes instance attributes which contain a list as a value.
    
    Assignments to this attributes, which must be dictionaries, are
    replaced by instances of 'dict_observer'.

    Note that you are not notified by changes to inner lists or maps.
    
    You should also be aware of aliasing.
    """
    def __init__ (self, external_attributename, observer):
        self.external_attributename = external_attributename 
        self.internal_attributename = '__'+external_attributename 
        self.observer = observer 
    
    def __set__ (self, instance, value):
        """Intercept assignments to the external attribute"""
        assert isinstance(value,type({}))
        if isinstance(value,dict_observer):
            newvalue = value 
            # if the value is already a dict_observer,
            # assume that the value is already monitored.
        else:
            newvalue = dict_observer(value, self.observer)
            internal_attributename = self.internal_attributename 
            try:
                oldvalue = getattr(instance, internal_attributename)
            except AttributeError:
                setattr(instance, self.internal_attributename, newvalue)
                self.observer.list_assignment_new(instance, internal_attributename)
            else:
                setattr(instance, self.internal_attributename, newvalue)
                self.observer.list_assignment_replace(instance, internal_attributename, oldvalue)
    
    def __get__ (self, instance, owner):
        try:
            return instance.__dict__[self.internal_attributename]
        except KeyError:
            return instance.__dict__[self.external_attributename]
    
if __name__ == '__main__':
    # minimal demonstration of the observer pattern.
    class c(object):
        pass
    observer = printargs()
    i = c()
    c.s = scalar_observer("x", observer)
    i.s = 1
    i.s = "hello"
    c.l = list_observer_in_instance("l", observer)
    i.l = [1, 2, 3]
    i.l.append(1)

The observer pattern is often used in loosely coupled system. Often, observers can dynamically subscribe and unsubscribe to the observed object. This implementation, however, only supports one observer.

The observer is send additional information about the change, so that the observer can undo the change, if desired.

This recipe builds on recipe http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/306864 to provide some limited support to observe changes to lists and dictionaries, if they are assigned to attributes of new style classes. See the inline documentation of the recipe.

This recipe is the second of two support recipes for the basic undo mechanism to be published later.