有没有办法将回调与列表的更改相关联?



例如,在jQuery中,你可以做

$(input).on("change", callback);

有没有办法在 Python 中做类似的事情,例如列表?像这样的东西(伪代码(:

from watch import Watchman
enemies = ["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"]
owl = Watchman()
def enemies_changed(old, new):
print(f"Enemies was {old}, now are {new}")
owl.watch(enemies, enemies_changed)
enemies.append("Alien")
# Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon'], now are ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']

方法 1:创建自己的类

您可以使用生成器和子类化来实现这一点list. 您必须覆盖要观看的每种方法。 下面是一个简单的示例。

def watcher(name=''):
while True:
x = yield
msg = f'{name} was {x}'
y = yield
if y is not None:
msg += f', now is {y}'
print(msg)

class List(list):
def __init__(self, *args, gen=None, **kwargs):
self.__gen = gen
next(gen)
super().__init__(*args, **kwargs)

def __add__(self, other):
try:
self.__gen.send(self)
super().__add__(other)
self.__gen.send(self)
except:
next(self.__gen)
raise

def __setitem__(self, *args, **kwargs):
try:
self.__gen.send(self)
super().__setitem__(*args, **kwargs)
self.__gen.send(self)
except:
next(self.__gen)
raise

def append(self, value):
self.__gen.send(self)
super().append(value)
self.__gen.send(self)

例子:

owl = watcher('Enemies')
enemies = List(["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"], gen=owl)
enemies.append('Alien')
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon'], now is ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']
enemies[-1] = 'Aliens'
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien'], now is ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Aliens']

方法2:猴子修补list

下面是一个猴子修补list内置类型以添加包装方法的示例。 在这种情况下,我们只是添加改变列表的方法的大写版本;即.append变成.追加'。

猴子补丁的代码来自 https://gist.github.com/bricef/1b0389ee89bd5b55113c7f3f3d6394ae。 您可以将其复制到名为patch.py的文件并使用from patch import monkey_patch_list

import ctypes
from types import MappingProxyType, MethodType

# figure out side of _Py_ssize_t
if hasattr(ctypes.pythonapi, 'Py_InitModule4_64'):
_Py_ssize_t = ctypes.c_int64
else:
_Py_ssize_t = ctypes.c_int

# regular python
class _PyObject(ctypes.Structure):
pass
_PyObject._fields_ = [
('ob_refcnt', _Py_ssize_t),
('ob_type', ctypes.POINTER(_PyObject))
]

# python with trace
if object.__basicsize__ != ctypes.sizeof(_PyObject):
class _PyObject(ctypes.Structure):
pass
_PyObject._fields_ = [
('_ob_next', ctypes.POINTER(_PyObject)),
('_ob_prev', ctypes.POINTER(_PyObject)),
('ob_refcnt', _Py_ssize_t),
('ob_type', ctypes.POINTER(_PyObject))
]

class _DictProxy(_PyObject):
_fields_ = [('dict', ctypes.POINTER(_PyObject))]

def reveal_dict(proxy):
if not isinstance(proxy, MappingProxyType):
raise TypeError('dictproxy expected')
dp = _DictProxy.from_address(id(proxy))
ns = {}
ctypes.pythonapi.PyDict_SetItem(ctypes.py_object(ns),
ctypes.py_object(None),
dp.dict)
return ns[None]

def get_class_dict(cls):
d = getattr(cls, '__dict__', None)
if d is None:
raise TypeError('given class does not have a dictionary')
if isinstance(d, MappingProxyType):
return reveal_dict(d)
return d

class Listener:
def __init__(self):
self._g = None
def __call__(self, x=None):
if x is None:
return self._g
self._g = x
def send(self, val):
if self._g:
self._g.send(val)

def monkey_patch_list(decorator, mutators=None):
if not mutators:
mutators = (
'append', 'clear', 'extend', 'insert', 'pop', 'remove', 
'reverse', 'sort'
)
d_list = get_class_dict(list)
d_list['_listener'] = Listener()
for m in mutators:
d_list[m.capitalize()] = decorator(d_list.get(m))

现在我们基本上可以使用上面的内容,即定义一个装饰器,它包装列表方法并在突变之前和突变后再次显示列表的str表示。 然后,可以传递接受两个参数和一个name关键字参数的任何函数来处理警报。

def before_after(clsm):
'''decorator for list class methods'''
def wrapper(self, *args, **kwargs):
self._listener.send(self)
out = clsm(self, *args, **kwargs)
self._listener.send(self)
return out
return wrapper

class Watchman:
def __init__(self):
self.guarding = []
def watch(self, lst, fn, name='list'):
self.guarding.append((lst, name))
w = self._watcher(fn, name)
lst._listener(w)

@staticmethod
def _watcher(fn, name):
def gen():
while True:
x = yield
x = str(x)
y = yield
y = str(y)
print(fn(x, y, name=name))
g = gen()
next(g)
return g

现在,您可以通过标准列表构造函数修补和使用新方法。

def enemies_changed(old, new, name='list'):
print(f"{name} was {old}, now are {new}")

# update the list methods with the wrapper
monkey_patch_list(before_after)
enemies = ["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"]
owl = Watchman()
owl.watch(enemies, enemies_changed, 'Enemies')
enemies.Append('Alien')
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon'], 
now are ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']

方法 3:使用自己的实现重载内置list

此方法类似于方法 1,但您仍然可以像往常一样使用list构造函数。 我们基本上会用我们自己的子类化版本覆盖内置的list类。 这些方法是相同的语法,我们只是添加一个侦听器和接收器属性,我们还包装了变异方法,以便侦听器拾取它们并向接收器发送信号是设置的。

# save the built-in list
_list = list
class list(_list):
def __init__(self, *args, emit_change=False, **kwargs):
super().__init__(*args, **kwargs)
self._emit = emit_change
self._listener = self._make_gen() if emit_change else None
self._init_change_emitter()
self._receiver = None
@property
def emit_change(self):
return self._emit

@property
def emitter(self):
return self._emitter
def _make_gen(self):
def gen():
while True:
x = yield
x = str(x)
y = yield
y = str(y)
yield (x, y)
g = gen()
next(g)
return g

def _init_change_emitter(self):
def before_after(clsm):
def wrapper(*args, **kwargs):
if self._listener:
self._listener.send(self)
out = clsm(*args, **kwargs)
before, after = self._listener.send(self)
if self._receiver:
self._receiver.send((before, after))
else:
out = clsm(*args, **kwargs)
return out
return wrapper
mutators = (
'append', 'clear', 'extend', 'insert', 'pop', 'remove',
'reverse', 'sort'
)
for m_str in mutators:
m = self.__getattribute__(m_str)
self.__setattr__(m_str, before_after(m))

在这一点上,list就像以前一样工作。 如您所料,使用list('abc')返回输出['a', 'b', 'c']list('abc', emit_change=True)也是如此。 额外的关键字参数允许钩住接收器,当它发生变化时,在列表片段之前和之后发送。

使用括号构造的列表需要通过list构造函数传递才能打开侦听/发射。

例:

class Watchman:
def __init__(self):
self.guarding = []

def watch(self, lst, fn, name='list'):
if not lst._listener:
raise ValueError(
'Can only watch lists initialized with `emit_change=True`.'
)
r = self._make_receiver(fn, name)
lst._receiver = r
self.guarding.append((lst, name, r))

def _make_receiver(self, fn, name):
def receiver():
while True:
x, y = yield
print(fn(x, y, name=name))
r = receiver()
next(r)
return r

def enemies_changed(old, new, name):
return f"{name} was {old}nNow is {new}"


enemies = ["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"]
enemies = list(enemies, emit_change=True)
owl = Watchman()
owl.watch(enemies, enemies_changed, 'Enemies')
enemies.append('Alien')
# prints:
Enemies was: ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon']
Now is: ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']

希望这些帮助之一!

最新更新