python - 有没有办法获取添加到 SQLAlchemy 模型的所有自定义事件监听器的列表？

Question

假设我已经向模型添加了一些事件监听器，并且我想获取模型的所有这些添加事件的列表，以在使用断言进行测试期间验证它们的存在。有办法吗？

我知道 SQLAlchemy 的 inspect ，我目前使用它来断言列和关系的存在。但是，有没有一种方法也可以通过 inspect 获取自定义事件监听器列表？如果没有，还有其他方法吗？我只想获取已明确添加到模型中的事件，而不是默认存在的事件(如果可能)。

我希望如何检索事件监听器的示例:

def test_schema(self):
    # sanity checks
    # this will raise any flags in the event schema is modified, so we know to update the appropriate tests
    assert tuple(inspect(MyModel).columns.keys()) == (
        "id", "module", "slug", "display_name"
    )
    assert tuple(inspect(MyModel).relationships.keys()) == ("accounts", "reports", "jobs")
    assert tuple(inspect(MyModel).events) == (
        "{event_function_name}_{trigger_action}",
        "{notify_manager_of_billing_changes}_{after_update}"
    )

def notify_manager_of_billing_changes(mapper, connection, model_instance):
    print(model_instance.billing_address)


from sqlalchemy import event
event.listen(MyModel, "after_update", notify_manager_of_billing_changes, retval=False)

Answer 1

测试特定的事件监听器

此类测试的公共(public) API 是:

assert event.contains(MyModel, "after_update", notify_manager_of_billing_changes)

获取所有自定义事件监听器的列表

SQLAlchemy 不跟踪函数名，只跟踪它的id¹ 和一个wrap 函数²。
¹ 在 id(notify_manager_of_billing_changes) 中。
² 不使用 functools.wraps!

在 call_function_get_frame 的帮助下来自答案 How can I get the values of the locals of a function ，添加一个except IndexError:，我们可以从wrap函数中获取对fn的引用。

import sys

from sqlalchemy.orm import Mapper


def call_function_get_frame(func, *args, **kwargs):
    """
    Calls the function *func* with the specified arguments and keyword
    arguments and snatches its local frame before it actually executes.
    """
    frame = None
    trace = sys.gettrace()
    def snatch_locals(_frame, name, arg):
        nonlocal frame
        if frame is None and name == 'call':
            frame = _frame
            sys.settrace(trace)
        return trace
    sys.settrace(snatch_locals)
    try:
        result = func(*args, **kwargs)
    except IndexError:  # Added
        result = None   # Added
    finally:
        sys.settrace(trace)
    return frame, result


def get_events(mapper):
    events = []
    dispatch = mapper.dispatch
    for event_name in dispatch._event_names:
        listeners = getattr(dispatch, event_name).listeners
        for wrap in listeners:
            frame, result = call_function_get_frame(wrap)
            events.append(f"{{{frame.f_locals['fn'].__name__}}}_{{{event_name}}}")
    return events


Mapper.events = property(get_events)

用法，根据问题的需要:

assert tuple(inspect(MyModel).events) == (
    # "{event_function_name}_{trigger_action}",
    "{notify_manager_of_billing_changes}_{after_update}",
)