How to monitor/log sqlalchemy scoped sessions?

I am working on a legacy application that uses SqlAlchemy and whose architecture is...well...suboptimal.

I recently discovered there is a lot of (unneeded) mysql connections openned by the application, and I would like to pin down the code that does that. I suppose they are the result of scoped_sessions left open.

I could search them manually but I am wondering if it is possible to instrumentalise the code to discover the bugged functions/modules. (A SQLAlchemy monitor would be useful, but I don't think that exists).


SessionExtension will be usefull for inspecting

For example:

import traceback
from collections import defaultdict
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.orm.interfaces import SessionExtension


# TODO cleanup commited sessions
class MySessionUsageInspector(SessionExtension):

    def __init__(self):
        self.connections = defaultdict(list)

    def after_begin(self, session, transaction, connection):
        self.connections[connection].append(traceback.format_stack()[0])

    def repr_usage(self):
        for cnn, callers in self.connections.items():
            print(cnn)
            for index, caller in enumerate(callers):
                print('\t', index, caller)


if __name__ == '__main__':
    engine = create_engine('sqlite://')
    session_inspector = MySessionUsageInspector()
    Session = scoped_session(sessionmaker(bind=engine, 
        extension=session_inspector)
    )
    session = Session()

    session.execute('select 1;')
    session.commit()
    session.execute('select 2;')

    print('Session usage:')
    session_inspector.repr_usage()