使用 sqlalchemy 实现“软删除”系统

Question

我们正在使用 tornado 和 sqlalchemy 为应用创建服务。该应用程序是用 django 编写的，并使用“软删除机制”。这意味着基础 mysql 表中没有删除。要将行标记为已删除，我们只需将属性“删除”设置为 True。但是，在服务中我们使用的是 sqlalchemy。最初，我们开始在通过 sqlalchemy 本身发出的查询中添加检查删除，例如：

customers = db.query(Customer).filter(not_(Customer.deleted)).all()

但是，这会导致很多潜在的错误，因为开发人员往往会错过在查询中删除的检查。因此，我们决定使用执行“预过滤器”的查询类覆盖默认查询：

class SafeDeleteMixin(Query):
    def __iter__(self):
        return Query.__iter__(self.deleted_filter())
    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.deleted_filter(), *ent)
    def deleted_filter(self):
        mzero = self._mapper_zero()
        if mzero is not None:
            crit = mzero.class_.deleted == False
            return self.enable_assertions(False).filter(crit)
        else:
            return self

这灵感来自此处关于 sqlalchemy 文档的解决方案：

https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/PreFilteredQuery

但是，我们仍然面临一些问题，例如我们同时进行过滤和更新，并且使用上面定义的这个查询类，更新在应用过滤器进行更新时不遵守delete=False 的标准。

db = CustomSession(with_deleted=False)()
result = db.query(Customer).filter(Customer.id == customer_id).update({Customer.last_active_time: last_active_time })

如何在 sqlalchemy 中实现“软删除”功能

Answer 1

我在这里做过类似的事情。我们做的有点不同，我们做了一个所有数据库访问都通过的服务层，有点像控制器，但只用于数据库访问，我们称之为 ResourceManager，它的灵感来自“领域驱动设计”（好书，对于很好地使用 SQLAlchemy 非常宝贵）。每个聚合根都存在一个派生的 ResourceManager，即。您想通过的每个资源类。 （虽然有时对于非常简单的 ResourceManagers，派生的管理器类本身是动态生成的）它有一个方法可以发出您的基本查询，并且该基本查询在分发之前会针对您的软删除进行过滤。从那时起，您可以生成地添加到该查询以进行过滤，最后使用 query.one() 或 first() 或 all() 或 count() 调用它。请注意，对于这种生成式查询处理，我遇到了一个问题，如果您加入一张表的次数过多，您可能会自缢。在某些情况下，为了过滤，我们必须跟踪哪些表已经被连接。如果您的删除过滤器不在主表中，只需先过滤它，然后您就可以随意加入。

所以是这样的：

class ResourceManager(object):
     # these will get filled in by the derived class
     # you could use ABC tools if you want, we don't bother
     model_class = None
     serializer_class = None

     # the resource manager gets instantiated once per request
     # and passed the current requests SQAlchemy session 
     def __init__(self, dbsession):
         self.dbs = dbsession 

     # hand out base query, assumes we have a boolean 'deleted' column
     @property
     def query(self):
         return self.dbs(self.model_class).filter(
            getattr(self.model_class, 'deleted')==False)

 class UserManager(ResourceManager):
     model_class = User

 # some client code might look this
 dbs = SomeSessionFactoryIHave()
 user_manager = UserManager(dbs)   
 users = user_manager.query.filter_by(name_last="Duncan").first()        

现在只要我总是从一个 ResourceManager 开始，它还有其他好处（参见前面提到的书），我知道我的查询是预先过滤的。这对我们当前具有软删除和相当广泛而棘手的数据库模式的项目非常有效。

hth！

Answer 2

我会创建一个函数

def customer_query():
    return db.session.query(Customer).filter(Customer.deleted == False)

我使用查询函数来不忘记默认标志，根据用户权限设置标志，使用连接过滤等，这样这些东西就不会被复制粘贴和忘记在各个地方。