【问题标题】:Is there a way to associate a callback to the change of a list?有没有办法将回调与列表的更改相关联?
【发布时间】:2020-05-17 21:37:35
【问题描述】:

以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']

【问题讨论】:

标签: python events observable watch


【解决方案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']

方法二:猴子补丁list

这是一个猴子修补 list 内置类型以添加​​包装方法的示例。在这种情况下,我们只是添加改变列表的方法的大写版本;即.append 变为 .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内置

此方法与方法一类似,但您仍然可以照常使用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) 也是如此。附加的关键字参数允许挂钩在列表的 sn-ps 之前和之后发送的接收器,当它发生变异时。

使用方括号构造的列表需要通过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']

希望其中一项有所帮助!

【讨论】:

  • Mh.不,谢谢,我宁愿不做子类。不过还是谢谢你的回答:-)
  • 对于默认 list 类的新方法中的猴子补丁,您可以接受吗?
  • 好吧,我又加了2个方法,看看#3是不是特别有用
猜你喜欢
  • 2021-03-13
  • 1970-01-01
  • 2022-08-20
  • 2012-08-17
  • 1970-01-01
  • 1970-01-01
  • 2011-02-08
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多