如何監視/記錄sqlalchemy範圍會話？

Question

我正在使用使用SqlAlchemy的遺留應用程序，其架構是......良好...不理想。

我最近發現有很多（不需要的）mysql連接被應用程序打開，我想確定那些代碼。我想他們是scoped_sessions保持打開狀態的結果。

我可以手動搜索他們，但我想知道是否有可能instrumentalise代碼來發現bug的功能/模塊。 （SQLAlchemy監視器會很有用，但我認爲不存在）。

Answer 1

SessionExtension將是有益的檢查

例如：

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

對不起，我的英語不好（