python之如何监控/记录 sqlalchemy 范围内的 session

我正在开发一个使用 SqlAlchemy 的遗留应用程序，它的体系结构……好吧……不是最理想的。

我最近发现应用程序打开了很多(不需要的)mysql 连接，我想确定执行此操作的代码。我想它们是 scoped_sessions 保持打开状态的结果。

我可以手动搜索它们，但我想知道是否可以使用代码来发现有问题的函数/模块。 (SQLAlchemy 监视器会很有用，但我认为它不存在)。

请您参考如下方法:

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