在 SQLAlchemy 中拦截模型上的所有查询

Question

我需要拦截所有与 SQLAlchemy 中的模型相关的查询，以便在任何查询方法（all()、one()、scalar() 等）所在的位置检查它执行。

我考虑过以下方法：

1。子类化查询类

我可以继承sqlalchemy.orm.Query 并覆盖执行代码，基本上从this 之类的东西开始。

但是，我正在编写一个可以在其他 SQLAlchemy 应用程序中使用的库，因此创建声明性基础（更不用说引擎和会话）超出了我的范围。

也许我错过了一些东西，并且可以在不知道会话的情况下覆盖我的模型的 Query 类？

2。使用 before_execute 核心事件

我也想过用before_execute 事件挂钩执行。

问题在于它与引擎绑定（见上文）。另外，我需要修改会话中的对象，我的印象是我无法从该事件中访问会话。

---

我想要做的是：

1. session.query(MyModel).filter_by(foo="bar").all() 被执行。
2. 拦截该查询并执行诸如将查询存储在同一数据库中的日志表中（不是字面意思，而是一组不同的东西，基本上需要与此示例操作完全相同的功能）
3. 让查询正常执行。

我最终要做的是在查询时将另一个数据存储中的项目即时注入 SQLAlchemy 数据库。虽然这看起来很愚蠢 - 相信我，它可能没有听起来那么愚蠢（甚至更愚蠢）;)。

Answer 1

before_compile 查询事件可能对您有用。

from weakref import WeakSet
from sqlalchemy import event
from sqlalchemy.orm import Query

visited_queries = WeakSet()

@event.listens_for(Query, 'before_compile')
def log_query(query):
    # You can get the session
    session = query.session

    # Prevent recursion if you want to compile the query to log it!
    if query not in visited_queries:
        visited_queries.add(query)
        # do something with query.statement

您可以查看query.column_descriptions 以查看您的模型是否正在被查询。