python - 如何监视/记录 sqlalchemy 范围的会话？

Question

我正在开发一个使用 SqlAlchemy 的遗留应用程序，其架构......嗯......次优。

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

我可以手动搜索它们，但我想知道是否可以利用代码来发现错误的功能/模块。（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()