python SQLAlchemy:一次为所有会话的查询给予一些execution_options参数

mm5n2pyu  于 2022-12-25  发布在  Python
关注(0)|答案(1)|浏览(142)

基于官方文档here中的示例,我在SQLAlchemy 1.4上设置了一个软删除,只要execute_state.execution_options"include_deleted"为False,_add_filtering_criteria就会过滤掉软删除的对象。
在某些情况下,我希望某些查询能够搜索软删除的对象。我可以针对每个查询执行此操作,即指定query.execution_options(include_deleted=True),但我希望特定会话的所有查询都包含软删除的对象,而不必针对每个查询指定它。
我尝试在引擎创建时声明execution_options(include_deleted=True),但它不起作用。

from sqlalchemy import orm

class SoftDeleted:
    deletion_date = Column(DateTime, nullable=True)

@event.listens_for(orm.Session, "do_orm_execute")
def _add_filtering_criteria(execute_state: orm.ORMExecuteState) -> None:
    """Intercepts all ORM queries. Adds a with_loader_criteria option to all
    of them.
    This option applies to SELECT queries and adds a global WHERE criteria
    (or as appropriate ON CLAUSE criteria for join targets)
    to all objects of a certain class or superclass.
    """
    if (not execute_state.is_column_load  
            and not execute_state.is_relationship_load
            and not execute_state.execution_options.get("include_deleted", False)):
        execute_state.statement = execute_state.statement.options(
            orm.with_loader_criteria(
                SoftDeleted,
                lambda cls: cls.deletion_date.is_(None),
                include_aliases=True,
            )
        )
engine = sa.create_engine(url, echo=False).execution_options(include_deleted=True)
session_factory = orm.sessionmaker(bind=engine)
session = session_factory()

# let SomeClass be a class that inherits from the SoftDeleted mixin

myquery = session.query(SomeClass).get(1) 
# will not retrieve SomeClass with uid=1 if it is soft-deleted

myquery2 = session.query(SomeClass).execution_options(include_deleted=True).get(1) 
# will retrieve SomeClass with uid=1 even if it is soft-deleted

正如我所说,我希望所有的查询会话能够包括软删除的对象。有人知道我怎么做吗?

bmp9r5qi

bmp9r5qi1#

您已在引擎的执行选项中设置了标志,因此必须从该选项中检索它。可以通过会话的bind属性访问引擎:

(Pdb) execute_state.session.bind.get_execution_options()
immutabledict({'include_deleted': True})

您的代码不会出错,因为execute_state具有execution_options属性,但它包含其他内容:

(Pdb) execute_state.execution_options
immutabledict({'_sa_orm_load_options': <class 'sqlalchemy.orm.context.QueryContext.default_load_options'>, '_result_disable_adapt_to_context': True, 'future_result': True})

会话对象(和sessionsmakers)有一个info属性,一个你可以随意填充的字典。如果你想为每个会话而不是每个引擎设置标志,这可以用来传递标志。

相关问题