使用_hint拦截并修改SQLAlchemy查询

Question

我有一个连接到 SQL Server 的 Flask + SQLAlchemy 应用程序。我想将

with_hint

方法注入到发出的每个查询中，以便我可以查询系统时态表。我想为每个请求定义一个当前时间戳

t

，然后修改每个查询以及这些查询中的所有表以包含

.with_hint(Model, 'FOR SYSTEM_TIME AS OF :t)

。

我想做一些类似于here描述的全局过滤器的事情，但我不知道如何适应它：

1. 识别查询中使用的每个模型
2. 将

   with_hint

   附加到每个模型
3. 将

   t

   参数注入到事件处理程序中的提示中

这仅与

SELECT

查询相关。无需处理

INSERT, UPDATE, DELETE

。

Answer 1

您确实可以使用事件挂钩来实现这一点。我没有正在运行的 MSSQL，但我想它应该是相同的，因为

with_hint

适用于所有方言：

from sqlalchemy.orm import mapped_column, relationship, DeclarativeBase, sessionmaker
from sqlalchemy import ForeignKey, Table, Column, select, Integer, create_engine, event
from sqlalchemy.sql.visitors import ClauseVisitor


# MODELS
class SubModel(Base):
    __tablename__ = "submodel"

    id = mapped_column(Integer, primary_key=True, nullable=False)


class Model(Base):
    __tablename__ = "model"

    id = mapped_column(Integer, primary_key=True, nullable=False)
    sub_model_id = mapped_column(Integer, ForeignKey(SubModel.id))
    sub_model = relationship(SubModel, uselist=False)

# DB config
engine = create_engine("mysql://...", echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

@event.listens_for(Session, "do_orm_execute")
def _do_orm_execute(orm_execute_state):
    if (
        orm_execute_state.is_select
        and not orm_execute_state.is_column_load
        and not orm_execute_state.is_relationship_load
    ):
        visitor = ClauseVisitor()
        for clause in visitor.iterate(orm_execute_state.statement):
            match clause:
                case Table():
                    orm_execute_state.statement = orm_execute_state.statement.with_hint(clause, "/*+ hint*/")
                case Column():
                    orm_execute_state.statement = orm_execute_state.statement.with_hint(clause.table, "/*+ hint*/")
                case _:
                    print("Did some tests but not 100% sure this covers everything")

# EXAMPLE

stmt = select(Model.id).join(Model.sub_model)
sess = Session()
sess.execute(stmt)
"""
> SELECT model.id 
FROM model /*+ hint*/ INNER JOIN submodel /*+ hint*/ ON submodel.id = model.sub_model_id
"""