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()