sqlalchemy - 将全局过滤器应用于 SQLAlchemy 中每个查询的所有表

Question

我们正在尝试在共享数据库和架构中设置支持 Multi-Tenancy 的 SaaS 服务。我们计划在所有表上都有一个tenant_id 列。我想要做的是，无需开发人员编写任何额外的代码，我的查询就可以通过此租户 ID 自动过滤所有涉及的表。在 SQL Alchemy 中是否有透明的方法来实现这一点？

我找到了如何覆盖默认查询对象:

self.session = sessionmaker(bind=engine, query_cls=TenantLimitingQuery)

但是在 TenantLimitingQuery 内部如何将其应用于所有涉及的表？

class TenantLimitingQuery(Query):
    def get(self, ident):
        #apply filter here

我的表有相同的列来标识租户，称为tenant_id，因此在该获取函数中，我需要按tenant_id=current_tenant_id进行过滤

Answer 1

这在 usage recipes wiki 中有概述。 ，转载在这里:

from sqlalchemy.orm.query import Query

class LimitingQuery(Query):

    def get(self, ident):
        # override get() so that the flag is always checked in the 
        # DB as opposed to pulling from the identity map. - this is optional.
        return Query.get(self.populate_existing(), ident)

    def __iter__(self):
        return Query.__iter__(self.private())

    def from_self(self, *ent):
        # override from_self() to automatically apply
        # the criterion too.   this works with count() and
        # others.
        return Query.from_self(self.private(), *ent)

    def private(self):
        mzero = self._mapper_zero()
        if mzero is not None:
            crit = mzero.class_.public == True

            return self.enable_assertions(False).filter(crit)
        else:
            return self

这个想法是在查询对象被迭代时按需应用过滤器。

如果您还希望将过滤器应用于关系，则需要使用 this recipe反而。