How do I get SQLAlchemy objects after commit?

The Problem

I am using a pattern to cache SQLAlchemy objects in Redis. Whenever an instance is modified and committed, I want to clear the relevant caches so that it will be reloaded when next fetched. This clear must happen after commit to avoid race conditions (another thread querying cache, missing, and reloading stale data into the cache).

I have fought with this for a long time, coming up with various solutions that work sometimes but nothing bulletproof. This seems like a straightforward enough problem that there should be a solution to it. How do I trigger some code every time a change is committed to a SQLAlchemy instance?

What I've Tried

Events

I've tried to stitch together some SQLAlchemy events to achieve my goal with varying levels of success. Listening to "after_insert" and "after_update" will tell me when an object is modified, and "after_commit" tells me that whatever was modified was saved, so I had a scheme where the first two events would register listeners for "after_commit", which would in turn pass the object to my cache clearing function. Like this:

def _register_after_commit(_: Mapper, __: Connection, target: MyClass) -> None:
    """ Generic callback that adds this function for a target change without params """
    targets.add(target)  # Clear cache uses this set to know which instances to clear
    event.listen(get_session(), "after_commit", clear_cache)

event.listen(MyClass, "after_insert", _register_after_commit)
event.listen(MyClass, "after_update", _register_after_commit)

This works most of the time, but I occasionally get DetachedInstanceError when accessing the attributes on the target that I need to know to clear them from cache (e.g. id). I've read that this happens because of automatic expiring during the commit which causes SQLAlchemy to want to refresh all attributes. I can't turn off auto-expiring nor can I expunge every object that passes through here, either of those could end up breaking other pieces of the code base.

A Custom Session

I made my own session class that looked something like this:

class SessionWithCallback(scoped_session):
    """ A version of orm.Session which can call a method after commit completes """

    def __init__(self, session_factory, scopefunc = None) -> None:
        super().__init__(session_factory=session_factory, scopefunc=scopefunc)
        self._callbacks = {}

    def add_callback(self, func, *args, **kwargs) -> None:
        """
        Adds a callback to be called after commit, ensuring only a single
        instance of the callback for each set of args and kwargs is used
        """
        key = f"{func}.{args}.{kwargs}"
        self._callbacks[key] = (func, args, kwargs)

    def run_callbacks(self) -> None:
        """
        Executes all callbacks
        """
        for (func, args, kwargs) in self._callbacks.values():
            func(*args, **kwargs)
        self._callbacks = {}

    def commit(self) -> None:
        """ Flush and commit the current transaction """
        super().commit()
        self.run_callbacks()

Then instead of _register_after_commit using the "after_commit" event, it would call the current session's add_callback function. This seemed to work when running tests with just SQLAlchemy, but it fell apart when integrating with a Flask app which uses these models and utilizes Flask-SQLAlchemy. I followed instructions to customize session (overriding create_session on the SQLAlchemy instance) but as soon as I commit anything I get an exception that scoped_session has no attribute add_callback. I stepped through and it is using my class internally somehow, but the session it gives me is not an instance of my class. Confusing.

I've Considered

1. Storing primary keys in my listeners, then requiring the callbacks to open a session and query for a new instance itself if it needs more info. Might work, but it feels like extra I/O that I shouldn't need. I could have several different callbacks for one instance, them all querying feels like a lot of work.
2. Having some global place to store callbacks instead of on the Session, so that I can avoid the add_callback function. I'd need to still make this session-specific and thread-safe though. Easy enough in Flask, but Flask isn't the only app that needs to share this code.
3. Just doing these cache clears manually... but that's bound to cause developer error.
4. Spawning some time-delayed job to clear cache from "after_insert/update". That gets really complicated really fast and sounds like a real headache. For instance, how do you decide how long to wait?