Есть ли способ связать обратный вызов с изменением списка? - PullRequest
1 голос
/ 01 февраля 2020

В 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 Ответ

1 голос
/ 01 февраля 2020

Метод 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 становится .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']

Надеюсь, одна из этих подсказок поможет!

...