为Python属性添加观察者模式装饰器的实现问题

问题根源 #

1. 返回描述器实例而非属性值：你的observable类__get__方法直接返回自身，导致opt.debug得到的是observable对象而非实际布尔值，无法直接当作布尔类型使用。
2. 观察者列表类级别共享：observers存储在observable类属性中，所有Options实例会共享同一组观察者，不同实例的观察者会互相干扰。
3. IDE遮蔽错误：IDE识别到你用自定义装饰器替换了原debug方法，且该装饰器结构与标准@property不一致，因此提示“遮蔽”错误——它认为原方法被覆盖，但无法识别新的属性结构。

修正后的代码实现 #

from typing import Callable, TypeVar, Generic, Any

T = TypeVar('T')
_Obj = TypeVar('_Obj')

class observable(Generic[T]):
    def __init__(self, fget: Callable[[_Obj], T]):
        self.fget = fget
        self.fset: Callable[[_Obj, T], None] | None = None
        self.name = fget.__name__

    def add_observer(self, obj: _Obj, observer: Callable[[T], None]) -> None:
        # 将观察者存储在实例对象中，避免类级别共享
        if not hasattr(obj, '__observable_observers__'):
            obj.__observable_observers__ = {}
        # 按属性名分组观察者
        observer_list = obj.__observable_observers__.setdefault(self.name, [])
        observer_list.append(observer)

    def setter(self, fset: Callable[[_Obj, T], None]) -> 'observable[T]':
        self.fset = fset
        return self

    def __set__(self, obj: _Obj, value: T) -> None:
        if self.fset is None:
            raise AttributeError(f"Setter not defined for property {self.name}")
        # 更新属性值
        self.fset(obj, value)
        # 通知所有观察者
        if hasattr(obj, '__observable_observers__'):
            for observer in obj.__observable_observers__.get(self.name, []):
                observer(value)

    def __get__(self, obj: _Obj, objtype: type[_Obj] | None = None) -> Any:
        if obj is None:
            return self  # 类访问时返回描述器本身

        # 创建代理对象，既支持获取属性值，又支持添加观察者
        class ObservableProxy:
            def __init__(self, proxy_obj: _Obj, descriptor: 'observable[T]'):
                self._obj = proxy_obj
                self._descriptor = descriptor

            def add_observer(self, observer: Callable[[T], None]) -> None:
                self._descriptor.add_observer(self._obj, observer)

            # 实现魔术方法，让代理可以当作属性值使用
            def __bool__(self) -> bool:
                return bool(self._descriptor.fget(self._obj))

            def __str__(self) -> str:
                return str(self._descriptor.fget(self._obj))

            def __repr__(self) -> str:
                return repr(self._descriptor.fget(self._obj))

            def __eq__(self, other: object) -> bool:
                return self._descriptor.fget(self._obj) == other

            def __ne__(self, other: object) -> bool:
                return self._descriptor.fget(self._obj) != other

        return ObservableProxy(obj, self)


class Options:
    def __init__(self):
        self._debug: bool = False

    @observable
    def debug(self) -> bool:
        return self._debug

    @debug.setter
    def debug(self, debug: bool):
        self._debug = debug


# 测试代码
opt = Options()
print(opt.debug)  # 输出: False
opt.debug.add_observer(lambda value: print(f"Debug is {value}"))
opt.debug = True  # 触发观察者，输出: Debug is True
print(opt.debug)  # 输出: True
if opt.debug:
    print("Debug mode is active")  # 输出: Debug mode is active
opt.debug = False  # 触发观察者，输出: Debug is False

关键改进点 #

1. 实例级观察者存储：将观察者列表存储在每个Options实例的__observable_observers__字典中，确保不同实例的观察者相互独立。
2. 代理对象实现双功能：通过ObservableProxy类，让opt.debug既可以当作布尔值使用（通过__bool__、__eq__等魔术方法），又能调用add_observer方法绑定观察者。
3. 兼容IDE类型识别：添加泛型类型注解，让IDE能更好地识别属性的实际类型；同时模仿@property的setter接口结构，让IDE更容易理解装饰器的作用，消除遮蔽错误提示。

内容的提问来源于stack exchange，提问作者Ignacio Mariotti