Is there a way to get a list of all custom event listeners added to a SQLAlchemy model?

Question

Say I've added a few event listeners to a model and I wanted to get a list of all of these added events for the model to verify their existence during testing with assertions. Is there a way to do so?

I'm aware of SQLAlchemy's inspect, which I currently use to assert the presence of columns and relationships. But, is there a way to obtain the list of custom event listeners through inspect as well? If not, is there any other way of doing so? I'd like to obtain only the events that have been added to the models explicitly, not those that are present by default (if possible).

Example of how I expect to retrieve event listeners:

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)

Should your tests not check that these work, as opposed to just exist? — Peter, Aug 04 '21 at 13:00
Yes, but in terms of a bare-bones unit test before moving onto functional tests I was curious if there is a way to accomplish this. Call it educational curiosity. — CaffeinatedMike, Aug 04 '21 at 13:50
@Peter For what its worth, I think this is a perfectly valid question. You shouldn't need to test that a listener works in the context of the sqlalchemy library, that's the libraries job and I think safe to assume that is well tested within sqlalchemy itself. In application code there are two possible sources of issues, 1. the logic of the handlers, and 2. the registration of the handlers. If there is a way to unittest that the the handler has been successfully registered without having to execute through layers of library code, what is the problem? — SuperShoot, Aug 05 '21 at 22:27

aaron · Accepted Answer · 2021-10-15T17:46:27.820

Testing specific event listeners

The public API for such a test is:

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

Getting a list of all custom event listeners

SQLAlchemy doesn't track the function name, only its id¹ and a wrap function².
¹ as in id(notify_manager_of_billing_changes).
² without using functools.wraps!

With the help of call_function_get_frame from the answer How can I get the values of the locals of a function, adding an except IndexError:, we can get the reference to fn from the wrap function.

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)

Usage, as desired in the question:

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

Is there a way to get a list of all custom event listeners added to a SQLAlchemy model?

1 Answers1

Testing specific event listeners

Getting a list of all custom event listeners