python - 通用对象 "watcher"代理，用于包装集合对象以记录状态更改(添加/删除)

Question

我想将“观察者”添加到库使用的一些变量中。 它们的类型为collections.deque、dict、weakref.WeakSet、 和列表。

每当向其中添加/附加或弹出项目时，我都想生成 一条日志消息。

我想尽量减少对原始代码的更改:

• 我不想在所有需要的地方添加日志消息 访问和修改变量。

• 我不想创建自己的继承自原始类的类 类(例如 class InstrumentedDeque(collections.deque): ...)

相反，这就是问题，是否有可能创建一个单一的 适用于所有集合的通用类(或包装器/装饰器) 上面的对象，因此唯一需要更改的地方是 对象是最初创建的。如果这是原始代码，则 2 “watch”的变量:self.scheduled 和 self.ready...

def __init__(self):
    self.scheduled = []
    self.ready = collections.deque()

那么唯一需要的改变就是......

def __init__(self):
    self.scheduled = MyLogger([], var_name='scheduled')
    self.ready = MyLogger(collections.deque(), var_name='ready')

没有仪器

test = Test()

test.add(1)
test.add(2)
test.pop()

仪器更改后

test = Test()

test.add(1)
***instrumentation output. scheduled: [1]
***instrumentation output. ready: deque([1])

test.add(2)
***instrumentation output. scheduled: [1, 2]
***instrumentation output. ready = deque([1, 2])

test.pop()
***instrumentation output. scheduled: [2]
***instrumentation output. ready: deque([2])

示例 add() 和 pop() 看起来像这样......

    def add(self, val):
        heapq.heappush(self.scheduled, val)
        self.ready.append(val)

    def pop(self):
        heapq.heappop(self.scheduled)
        self.ready.popleft()

我尝试创建一个“包装”类并使用 __new__、__init__、__getattr__ 但一直无法 让它发挥作用。像这样的事情...

class MyLooger:
    def __new__(cls):
        # what to do?

    def __init__(self):
        # what to do?

    def __getattr__(self, name):
        # what to do?

非常感谢任何帮助。

Answer 1

注意:以下抽象不适用于直接进入包装对象低级内部的 C 扩展代码(例如 heapq. heappush，都在 CPython 和 PyPy 上)；在 Python 层面上没有什么办法可以缓解这个问题。您可能看看是否可以在 C 级别“修补漏洞”，但随后您将不得不亲自动手编写 C 和 Python 扩展。

解决方案:您不需要走到__new__。以下内容通常适用于所有对象。它还将使 isinstance 在包装器上工作，就像在包装对象上调用它一样。

from functools import wraps

class Logged(object):
    def __init__(self, obj, obj_name):
        self.obj = obj
        self.obj_name = obj_name

    def __getattribute__(self, attr_name):
        obj = object.__getattribute__(self, 'obj')
        obj_name = object.__getattribute__(self, 'obj_name')

        attr = getattr(obj, attr_name)
        # this is not 100% generic, mathematically speaking,
        # but covers all methods and the `__class__` attribute:
        if not callable(attr) or isinstance(attr, type):
            return attr

        @wraps(attr)
        def fn(*args, **kwargs):
            print "%s called on %s with: %s and %s" % (attr_name, obj_name, args, kwargs)
            return attr(*args, **kwargs)

        return fn

    def __repr__(self):
        return repr(object.__getattribute__(self, 'obj'))

然后就是:

>>> scheduled = Logged([], obj_name="scheduled")

>>> scheduled.append
<function append>

>>> scheduled.append(3)
append called on scheduled with: (3,) and {}

>>> scheduled.extend([1,2])
extend called on scheduled with: ([1, 2],) and {}

>>> isinstance(scheduled, list)
True

>>> scheduled
[3, 1, 2]