为何Python装饰器类的__get__方法未在所有场景下被调用？

问题描述 #

尝试实现类似C# events的Python方法装饰器，代码如下：

from __future__ import annotations
from typing import *
import functools


def event(function: Callable) -> EventDispatcher:
    return EventDispatcher(function)


class EventDispatcher:

    def __init__(self, function: Callable) -> None:
        functools.update_wrapper(self, function)

        self._instance: Any | None = None
        self._function: Callable = function
        self._callbacks: Set[Callable] = set()

    def __get__(self, instance: Any, _: Any) -> EventDispatcher:
        self._instance = instance
        return self

    def __iadd__(self, callback: Callable) -> EventDispatcher:
        self._callbacks.add(callback)
        return self

    def __isub__(self, callback: Callable) -> EventDispatcher:
        self._callbacks.remove(callback)
        return self

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        for callback in self._callbacks:
            callback(*args, **kwargs)

        return self._function(self._instance, *args, **kwargs)

正常场景：未初始化B实例时，调用不同A实例的my_event方法，输出不同实例地址，符合预期。

class A:
    @event
    def my_event(self) -> None:
        print(self)

class B:
    def __init__(self, a: A) -> None:
        a.my_event += self.my_callback
    def my_callback(self) -> None:
        pass

a0 = A()
a1 = A()
# b = B(a0)
a1.my_event()
a0.my_event()

输出：

<__main__.A object at 0x00000170AA15FCA0>
<__main__.A object at 0x00000170AA15FC70>

异常场景：取消注释b = B(a0)后，调用a1.my_event()和a0.my_event()输出同一个实例地址，不符合预期。

输出：

<__main__.A object at 0x000002067650FCA0>
<__main__.A object at 0x000002067650FCA0>

原因分析 #

你提到的__get__方法确实被调用了，但设计逻辑存在核心错误：

1. 共享实例状态：EventDispatcher作为类属性，所有A实例共享同一个EventDispatcher对象。当执行a0.my_event += callback时：

   • 触发__get__，将共享对象的_instance设为a0；
   • 由于__iadd__返回自身，Python会将这个共享对象赋值给a0实例的my_event属性（因为描述符未实现__set__，实例属性会覆盖类描述符）。

2. 状态被覆盖：后续访问a1.my_event时，触发类描述符的__get__，将同一个共享EventDispatcher对象的_instance改为a1。此时调用a0.my_event()时，使用的仍是这个共享对象，它的_instance已被最后一次访问的实例覆盖，导致所有调用输出同一个实例。

此外，所有实例还共享同一个_callbacks集合，这会导致一个实例添加的回调会影响其他实例，不符合C#事件的实例独立性要求。

修复方案 #

需要让每个类实例拥有独立的事件状态（回调集合、绑定实例），修改后的实现如下：

from __future__ import annotations
from typing import *
import functools


def event(function: Callable) -> EventDispatcher:
    return EventDispatcher(function)


class EventDispatcher:
    """类级别的描述符，负责为每个实例创建独立的事件绑定"""
    def __init__(self, function: Callable) -> None:
        functools.update_wrapper(self, function)
        self._function: Callable = function
        # 存储每个实例的事件数据：key为实例对象，value为包含回调和绑定方法的字典
        self._instance_data: Dict[Any, Dict[str, Any]] = {}

    def __get__(self, instance: Any, owner: type) -> BoundEvent | EventDispatcher:
        if instance is None:
            # 类级别的访问（如A.my_event），返回描述符自身
            return self
        # 为实例初始化专属的事件数据（如果不存在）
        if instance not in self._instance_data:
            self._instance_data[instance] = {
                'callbacks': set(),
                # 预先绑定实例到原方法，避免每次调用都手动传参
                'bound_func': functools.partial(self._function, instance)
            }
        # 返回绑定当前实例的事件对象
        return BoundEvent(instance, self._instance_data[instance])


class BoundEvent:
    """绑定到具体实例的事件对象，负责回调管理和事件触发"""
    def __init__(self, instance: Any, data: Dict[str, Any]) -> None:
        self._instance = instance
        self._callbacks = data['callbacks']
        self._bound_func = data['bound_func']

    def __iadd__(self, callback: Callable) -> BoundEvent:
        self._callbacks.add(callback)
        return self

    def __isub__(self, callback: Callable) -> BoundEvent:
        self._callbacks.remove(callback)
        return self

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        # 触发所有注册的回调
        for callback in self._callbacks:
            callback(*args, **kwargs)
        # 执行原方法
        return self._bound_func(*args, **kwargs)

修复逻辑说明 #

1. 分离类描述符与实例事件：EventDispatcher作为类级别的描述符，负责管理所有实例的事件数据，但不存储实例相关的状态。
2. 独立实例状态：每个实例的回调集合和绑定方法都存储在_instance_data字典中，实例之间完全隔离。
3. BoundEvent封装实例逻辑：BoundEvent专门处理单个实例的回调添加、移除和事件触发，确保每个实例的事件操作互不干扰。

测试修复后的代码，无论是正常场景还是异常场景，都能输出不同的实例地址，且每个实例的回调都是独立的。

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