使用 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 开始，它也有其他好处（参见上述书籍），我知道我的查询是预过滤的。这对我们当前的项目非常有效，该项目具有软删除和相当广泛且棘手的数据库架构。

嗯！