Python3 使用装饰器实现观察者模式
Using decorators to implement Observer Pattern in Python3
这道题一般不是关于观察者模式的。它专注于在该模式中使用装饰器。该问题基于类似问题的。
#!/usr/bin/env python3
class Observable:
"""
The object that need to be observed. Alternative names are 'Subject'.
In the most cases it is a data object.
"""
def __init__(self):
self._observers = []
def register_observer(self, callback):
self._observers.append(callback)
return callback
def _broadcast_observers(self, *args, **kwargs):
for callback in self._observers:
callback(*args, **kwargs)
class TheData(Observable):
"""
Example of a data class just for demonstration.
"""
def __init__(self, data):
Observable.__init__(self)
self._data = data
@property
def data(self):
return self._data
@data.setter
def data(self, data):
self._data = data
self._broadcast_observers()
class TheGUIElement:
"""
Example of a gui class (Widget) just for demonstration.
e. g. it could be a text field in GUI.
"""
def __init__(self, data):
self._data = data
#data.register_observer(self._data_updated)
self._redraw()
def _redraw(self):
print('in _redraw(): ' + data.data)
@Observable.register_observer
def _data_updated(self, **kwargs):
"""
This is the callback that is called by the Observable if the
data changed.
"""
print('in _data_updated() - kwargs: {}'.format(kwargs))
self._redraw()
if __name__ == '__main__':
data = TheData('DATA')
gui = TheGUIElement(data)
data.data = 'SECOND DATA'
由于此错误，此代码无法运行。
Traceback (most recent call last):
File "./o.py", line 42, in <module>
class TheGUIElement:
File "./o.py", line 55, in TheGUIElement
@Observable.register_observer
TypeError: register_observer() missing 1 required positional argument: 'callback'
我不清楚如何使用装饰器来注册观察者（例如 TheGUIElement）。
要注册回调，您需要有一个实际的对象。在您的代码中，@Observable.register_observer 应该如何找到应该注册的实例？
请放弃 Observable javaism 的东西，在 python 中很麻烦。
看看这个。
#!/usr/bin/env python
class SomeData(object):
def __init__(self, value):
self.callbacks = []
self.foo = value
def register(self, callback):
self.callbacks.append(callback)
return callback
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(self, *args, **kwargs)
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.register(my_gui.redraw)
# Try changing it. Note my_data is dumb for now, notify manually.
my_data.foo = 10
my_data.notify("foo", 10)
我特意删除了自动通知以说明注册本身。
让我们把它加回来。但是使用那个 Observable class 没有意义。让我们让它更轻，简单地定义一个事件 class.
#!/usr/bin/env python3
class Event(object):
def __init__(self):
self.callbacks = []
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(*args, **kwargs)
def register(self, callback):
self.callbacks.append(callback)
return callback
class SomeData(object):
def __init__(self, foo):
self.changed = Event()
self._foo = foo
@property
def foo(self):
return self._foo
@foo.setter
def foo(self, value):
self._foo = value
self.changed.notify(self, 'foo', value)
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.changed.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.changed.register(my_gui.redraw)
# Try changing it.
my_data.foo = 10
正如您现在可能注意到的，装饰器语法在这些情况下很有用：
- 您只有一个注册表。无论是单例还是class本身class都是一阶对象，大多数都是单例。
- 您动态定义函数并随时注册它。
现在，您的那些手册 getters/setters 也很麻烦，如果您有很多为什么不把它们分解出来呢？
#!/usr/bin/env python3
class Event(object):
def __init__(self):
self.callbacks = []
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(*args, **kwargs)
def register(self, callback):
self.callbacks.append(callback)
return callback
@classmethod
def watched_property(cls, event_name, key):
actual_key = '_%s' % key
def getter(obj):
return getattr(obj, actual_key)
def setter(obj, value):
event = getattr(obj, event_name)
setattr(obj, actual_key, value)
event.notify(obj, key, value)
return property(fget=getter, fset=setter)
class SomeData(object):
foo = Event.watched_property('changed', 'foo')
def __init__(self, foo):
self.changed = Event()
self.foo = foo
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.changed.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.changed.register(my_gui.redraw)
# Try changing it.
my_data.foo = 10
供参考，所有三个程序输出完全相同的内容：
$ python3 test.py
Key foo changed to 10
redrawing <__main__.SomeGUI object at 0x7f9a90d55fd0> with value 10
虽然帖子有点旧（可能问题已经解决了），但我想分享我对"Decorated Observer Pattern"问题的解决方案：
https://pypi.org/project/notifyr/
我创建了一个实现装饰器的包，该装饰器将观察者观察到的 methods/attributes 添加到 python 类。我也设法在 Django 项目中使用了该包，但进行了一些修改（.observers 属性未保留在数据库中，因此每次我希望通知他们时我都必须将观察者列表加载到其中）。
这是一个实现示例：
原码：
class Dog(object):
def __init__(self, name):
self.name = name
def bark(self):
print('Woof')
def sleep(self):
print(self.name, 'is now asleep: ZZzzzzZzzZ...')
class Person(object):
def __init__(self, name):
self.name = name
def educate_dog(self, dog):
print(self.name + ':','Sleep,', dog.name)
dog.sleep()
假设我们想要一个人在狗每次吠叫时教育它：
from notifyr.agents import observed, observer
from notifyr.functions import target
@observed
class Dog(object):
def __init__(self, name):
self.name = name
@target
def bark(self):
print('Woof')
def sleep(self):
print(self.name, 'is now asleep: ZZzzzzZzzZ...')
@observer('educate_dog')
class Person(object):
def __init__(self, name):
self.name = name
def educate_dog(self, dog):
print(self.name + ':','Sleep,', dog.name)
dog.sleep()
给定修饰后的类，可以得到如下结果：
d = Dog('Tobby')
p = Person('Victor')
d.attach(p) # Victor is now observing Tobby
d.bark()
# Woof
# Victor: Sleep, Tobby
# Tobby is now asleep: ZZzzzzZzzZ...
该软件包仍然非常原始，但它提供了针对此类情况的有效解决方案。
我最近在寻找类似的东西，这就是我想出的东西。它通过拦截 __setattr__ 方法来工作——这是一个有用的特技，我打算把它放在口袋里以备后用。
def watchableClass(cls):
"""
Class Decorator!
* If the class has a "dirty" member variable, then it will be
automatically set whenever any class value changes
* If the class has an "onChanged()" method, it will be called
automatically whenever any class value changes
* All this only takes place if the value is different from what it was
that is, if myObject.x is already 10 and you set myObject.x=10
nothing happens
* DOES NOT work with getter/setter functions. But then, you are
already in a function, so do what you want!
EXAMPLE:
@watchableClass
class MyClass:
def __init__(self):
self.dirty=False
def onChanged(self):
print('class has changed')
"""
if hasattr(cls,'__setattr__'):
cls.__setattr_unwatched__=cls.__setattr__
cls.__setattr__=_setObjValueWatchedCascade
else:
cls.__setattr__=_setObjValueWatched
return cls
def _setObjValueWatched(ob,k,v):
"""
called when an object value is set
"""
different=not k in ob.__dict__ or ob.__dict__[k]!=v
if different:
ob.__dict__[k]=v
if k not in ('dirty'):
if hasattr(ob,'dirty'):
ob.dirty=True
if hasattr(ob,'onChanged'):
ob.onChanged()
def _setObjValueWatchedCascade(ob,k,v):
"""
called when an object value is set
IF the class had its own __setattr__ member defined!
"""
different=not k in ob.__dict__ or ob.__dict__[k]!=v
ob.__setattr_unwatched__(k,v)
if different:
if k not in ('dirty'):
if hasattr(ob,'dirty'):
ob.dirty=True
if hasattr(ob,'onChanged'):
ob.onChanged()
这道题一般不是关于观察者模式的。它专注于在该模式中使用装饰器。该问题基于类似问题的
#!/usr/bin/env python3
class Observable:
"""
The object that need to be observed. Alternative names are 'Subject'.
In the most cases it is a data object.
"""
def __init__(self):
self._observers = []
def register_observer(self, callback):
self._observers.append(callback)
return callback
def _broadcast_observers(self, *args, **kwargs):
for callback in self._observers:
callback(*args, **kwargs)
class TheData(Observable):
"""
Example of a data class just for demonstration.
"""
def __init__(self, data):
Observable.__init__(self)
self._data = data
@property
def data(self):
return self._data
@data.setter
def data(self, data):
self._data = data
self._broadcast_observers()
class TheGUIElement:
"""
Example of a gui class (Widget) just for demonstration.
e. g. it could be a text field in GUI.
"""
def __init__(self, data):
self._data = data
#data.register_observer(self._data_updated)
self._redraw()
def _redraw(self):
print('in _redraw(): ' + data.data)
@Observable.register_observer
def _data_updated(self, **kwargs):
"""
This is the callback that is called by the Observable if the
data changed.
"""
print('in _data_updated() - kwargs: {}'.format(kwargs))
self._redraw()
if __name__ == '__main__':
data = TheData('DATA')
gui = TheGUIElement(data)
data.data = 'SECOND DATA'
由于此错误，此代码无法运行。
Traceback (most recent call last):
File "./o.py", line 42, in <module>
class TheGUIElement:
File "./o.py", line 55, in TheGUIElement
@Observable.register_observer
TypeError: register_observer() missing 1 required positional argument: 'callback'
我不清楚如何使用装饰器来注册观察者（例如 TheGUIElement）。
要注册回调，您需要有一个实际的对象。在您的代码中，@Observable.register_observer 应该如何找到应该注册的实例？
请放弃 Observable javaism 的东西，在 python 中很麻烦。
看看这个。
#!/usr/bin/env python
class SomeData(object):
def __init__(self, value):
self.callbacks = []
self.foo = value
def register(self, callback):
self.callbacks.append(callback)
return callback
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(self, *args, **kwargs)
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.register(my_gui.redraw)
# Try changing it. Note my_data is dumb for now, notify manually.
my_data.foo = 10
my_data.notify("foo", 10)
我特意删除了自动通知以说明注册本身。
让我们把它加回来。但是使用那个 Observable class 没有意义。让我们让它更轻，简单地定义一个事件 class.
#!/usr/bin/env python3
class Event(object):
def __init__(self):
self.callbacks = []
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(*args, **kwargs)
def register(self, callback):
self.callbacks.append(callback)
return callback
class SomeData(object):
def __init__(self, foo):
self.changed = Event()
self._foo = foo
@property
def foo(self):
return self._foo
@foo.setter
def foo(self, value):
self._foo = value
self.changed.notify(self, 'foo', value)
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.changed.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.changed.register(my_gui.redraw)
# Try changing it.
my_data.foo = 10
正如您现在可能注意到的，装饰器语法在这些情况下很有用：
- 您只有一个注册表。无论是单例还是class本身class都是一阶对象，大多数都是单例。
- 您动态定义函数并随时注册它。
现在，您的那些手册 getters/setters 也很麻烦，如果您有很多为什么不把它们分解出来呢？
#!/usr/bin/env python3
class Event(object):
def __init__(self):
self.callbacks = []
def notify(self, *args, **kwargs):
for callback in self.callbacks:
callback(*args, **kwargs)
def register(self, callback):
self.callbacks.append(callback)
return callback
@classmethod
def watched_property(cls, event_name, key):
actual_key = '_%s' % key
def getter(obj):
return getattr(obj, actual_key)
def setter(obj, value):
event = getattr(obj, event_name)
setattr(obj, actual_key, value)
event.notify(obj, key, value)
return property(fget=getter, fset=setter)
class SomeData(object):
foo = Event.watched_property('changed', 'foo')
def __init__(self, foo):
self.changed = Event()
self.foo = foo
class SomeGUI(object):
def redraw(self, obj, key, newvalue):
print('redrawing %s with value %s' % (self, newvalue))
if __name__ == '__main__':
my_data = SomeData(42)
# Register some function using decorator syntax
@my_data.changed.register
def print_it(obj, key, value):
print('Key %s changed to %s' % (key, value))
# Register the SomeGUI element
my_gui = SomeGUI()
my_data.changed.register(my_gui.redraw)
# Try changing it.
my_data.foo = 10
供参考，所有三个程序输出完全相同的内容：
$ python3 test.py
Key foo changed to 10
redrawing <__main__.SomeGUI object at 0x7f9a90d55fd0> with value 10
虽然帖子有点旧（可能问题已经解决了），但我想分享我对"Decorated Observer Pattern"问题的解决方案：
https://pypi.org/project/notifyr/
我创建了一个实现装饰器的包，该装饰器将观察者观察到的 methods/attributes 添加到 python 类。我也设法在 Django 项目中使用了该包，但进行了一些修改（.observers 属性未保留在数据库中，因此每次我希望通知他们时我都必须将观察者列表加载到其中）。
这是一个实现示例：
原码：
class Dog(object):
def __init__(self, name):
self.name = name
def bark(self):
print('Woof')
def sleep(self):
print(self.name, 'is now asleep: ZZzzzzZzzZ...')
class Person(object):
def __init__(self, name):
self.name = name
def educate_dog(self, dog):
print(self.name + ':','Sleep,', dog.name)
dog.sleep()
假设我们想要一个人在狗每次吠叫时教育它：
from notifyr.agents import observed, observer
from notifyr.functions import target
@observed
class Dog(object):
def __init__(self, name):
self.name = name
@target
def bark(self):
print('Woof')
def sleep(self):
print(self.name, 'is now asleep: ZZzzzzZzzZ...')
@observer('educate_dog')
class Person(object):
def __init__(self, name):
self.name = name
def educate_dog(self, dog):
print(self.name + ':','Sleep,', dog.name)
dog.sleep()
给定修饰后的类，可以得到如下结果：
d = Dog('Tobby')
p = Person('Victor')
d.attach(p) # Victor is now observing Tobby
d.bark()
# Woof
# Victor: Sleep, Tobby
# Tobby is now asleep: ZZzzzzZzzZ...
该软件包仍然非常原始，但它提供了针对此类情况的有效解决方案。
我最近在寻找类似的东西，这就是我想出的东西。它通过拦截 __setattr__ 方法来工作——这是一个有用的特技，我打算把它放在口袋里以备后用。
def watchableClass(cls):
"""
Class Decorator!
* If the class has a "dirty" member variable, then it will be
automatically set whenever any class value changes
* If the class has an "onChanged()" method, it will be called
automatically whenever any class value changes
* All this only takes place if the value is different from what it was
that is, if myObject.x is already 10 and you set myObject.x=10
nothing happens
* DOES NOT work with getter/setter functions. But then, you are
already in a function, so do what you want!
EXAMPLE:
@watchableClass
class MyClass:
def __init__(self):
self.dirty=False
def onChanged(self):
print('class has changed')
"""
if hasattr(cls,'__setattr__'):
cls.__setattr_unwatched__=cls.__setattr__
cls.__setattr__=_setObjValueWatchedCascade
else:
cls.__setattr__=_setObjValueWatched
return cls
def _setObjValueWatched(ob,k,v):
"""
called when an object value is set
"""
different=not k in ob.__dict__ or ob.__dict__[k]!=v
if different:
ob.__dict__[k]=v
if k not in ('dirty'):
if hasattr(ob,'dirty'):
ob.dirty=True
if hasattr(ob,'onChanged'):
ob.onChanged()
def _setObjValueWatchedCascade(ob,k,v):
"""
called when an object value is set
IF the class had its own __setattr__ member defined!
"""
different=not k in ob.__dict__ or ob.__dict__[k]!=v
ob.__setattr_unwatched__(k,v)
if different:
if k not in ('dirty'):
if hasattr(ob,'dirty'):
ob.dirty=True
if hasattr(ob,'onChanged'):
ob.onChanged()