如何在SQLAlchemy中检查事件侦听器的存在？

Question

我在SQLAlchemy映射类中实现版本控制，遵循示例代码here。你知道吗

下面是versioned_session函数，我在创建的每个会话中都会调用它：

def versioned_session(session):
    @event.listens_for(session, 'before_flush')
    def before_flush(session, flush_context, instances):
        for obj in versioned_objects(session.dirty):
            create_version(obj, session)
        for obj in versioned_objects(session.deleted):
            create_version(obj, session, deleted=True)

我的版本控制基本正常，除了刷新更改。在整个应用程序中，当创建会话时，before_flush事件侦听器被分配给会话。因为我使用的是scoped_session，所以应用程序似乎在重用同一个会话，但仍然在事件侦听器已经存在时添加它。这导致每次一个刷新即将发生时，都会创建大约20个相同对象的新版本。有没有一种方法可以让我有条件地注册事件侦听器，只在它还不存在的时候？

目前，我正在使用一种变通方法，添加一个自定义_has_versioning_listener标志。我觉得没有必要加上我自己的旗子。你知道吗

def versioned_session(session):
    if not getattr(session, '_has_versioning_listener', False):
        @event.listens_for(session, 'before_flush')
        def before_flush(session, flush_context, instances):
            for obj in versioned_objects(session.dirty):
                create_version(obj, session)
            for obj in versioned_objects(session.deleted):
                create_version(obj, session, deleted=True)
        session._has_versioning_listener = True