SQLAlchemy软删除怎么实现

wen python案例 1

SQLAlchemy软删除实践指南:从原理到完整实现方案

📖 文章导读

章节 内容说明
什么是软删除? 概念解析与核心价值
为什么需要软删除? 对比物理删除的5大优势
SQLAlchemy软删除实现方案 3种主流实现方式详解
完整代码示例 可复用的软删除Mixin
查询时自动过滤 如何让软删除透明化
软删除的陷阱与最佳实践 常见问题与规避策略
问答环节 5个高频实务问题解答

什么是软删除?

软删除(Soft Delete) 是指不真正从数据库中物理删除记录,而是通过标记(如is_deleted字段或deleted_at时间戳)来标识该记录已被“删除”,被标记的记录在常规查询中被过滤,但数据本身仍保留在表中。

SQLAlchemy软删除怎么实现

# 示例:软删除标记字段
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    is_deleted = Column(Boolean, default=False)  # 软删除标记
    deleted_at = Column(DateTime, nullable=True) # 删除时间

为什么需要软删除?

对比维度 物理删除 软删除
数据恢复 ❌ 需要binlog或备份 ✅ 直接取消标记
历史审计 ❌ 丢失记录 ✅ 完整保留轨迹
关联数据 ❌ 可能级联删除失误 ✅ 保留关联完整性
回滚操作 ❌ 不可逆 ✅ 可逆
查询性能 无影响(数据物理消失) 需添加过滤条件

核心价值:软删除是数据资产安全的最后一道防线,尤其适用于金融、医疗、电商等对数据追溯有严格要求的场景。


SQLAlchemy软删除实现方案

手动标记法(最基础)

class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    is_deleted = Column(Boolean, default=False)
    def soft_delete(self):
        self.is_deleted = True

缺点:每次查询需手动加过滤,容易遗漏。

使用@hybrid_property + 默认过滤器(推荐)

from sqlalchemy.ext.hybrid import hybrid_property
class SoftDeleteMixin:
    @hybrid_property
    def deleted(self):
        return self.is_deleted
    @deleted.expression
    def deleted(cls):
        return cls.is_deleted == True

结合Query事件拦截(专业级)

利用@event.listens_for监听查询,自动注入过滤条件。

from sqlalchemy import event
from sqlalchemy.orm import Query
class SoftDeleteQuery(Query):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._soft_delete_filter = True
    def get(self, ident):
        obj = super().get(ident)
        return obj if obj and not obj.is_deleted else None

完整代码示例:可复用的软删除Mixin

from sqlalchemy import Column, Boolean, DateTime, func
from sqlalchemy.ext.declarative import declared_attr
from datetime import datetime
class SoftDeleteMixin:
    """软删除基类混合器"""
    @declared_attr
    def is_deleted(cls):
        return Column(Boolean, default=False, nullable=False, 
                      comment='软删除标记')
    @declared_attr
    def deleted_at(cls):
        return Column(DateTime, nullable=True, 
                      comment='删除时间')
    # 实例方法
    def soft_delete(self):
        self.is_deleted = True
        self.deleted_at = datetime.utcnow()
    def restore(self):
        self.is_deleted = False
        self.deleted_at = None
    # 查询过滤器(配合自定义Query使用)
    @classmethod
    def not_deleted(cls, query):
        return query.filter(cls.is_deleted == False)
# 使用示例
class User(Base, SoftDeleteMixin):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
# 软删除
user = session.query(User).first()
user.soft_delete()
session.commit()
# 恢复
user.restore()
session.commit()

查询时自动过滤:让软删除透明化

重写Session的query方法(侵入性高)

class MySession(Session):
    def query(self, *entities, **kwargs):
        query = super().query(*entities, **kwargs)
        # 为所有带SoftDeleteMixin的实体添加过滤
        for entity in entities:
            if isinstance(entity, type) and SoftDeleteMixin in entity.__bases__:
                query = query.filter(entity.is_deleted == False)
        return query

使用SQLAlchemy 2.0的with_loader_criteria(推荐)

from sqlalchemy.orm import with_loader_criteria
for user in session.execute(
    select(User).options(
        with_loader_criteria(User, User.is_deleted == False)
    )
).scalars():
    print(user.name)

全局事件监听(零侵入)

@event.listens_for(Query, "before_compile", retval=True)
def soft_delete_filter(query):
    for entity in query.column_descriptions:
        entity_type = entity['type']
        if hasattr(entity_type, 'is_deleted'):
            query = query.filter(entity_type.is_deleted == False)
    return query

软删除的陷阱与最佳实践

⚠️ 常见陷阱

  1. 唯一约束冲突:软删除的记录仍会占用唯一键,如用户名。
    • 解决方案:使用复合唯一索引 (username, is_deleted) 或使用唯一删除标记
  2. 关联查询遗漏:外键关联未考虑软删除,导致级联问题。
    • 解决方案:在关联表上也添加软删除,或使用lazy='dynamic'自过滤
  3. 统计、报告数据偏差COUNT(*)包含软删除记录。
    • 解决方案:始终使用filter(is_deleted==False)

✅ 最佳实践清单

  • 使用deleted_at而非is_deleted,方便时间回溯
  • 为软删除字段建立索引(尤其表数据量大时)
  • 保留定时清理存储过程(如删除超过365天的软删除记录)
  • 在API层区分“删除”与“彻底删除”操作

问答环节

Q1:软删除与逻辑删除、伪删除是同一个概念吗? A:是的,在ORM语境下,软删除、逻辑删除、伪删除均指向同一模式——通过状态标记而非物理移除数据。

Q2:如何实现多人操作时不互相覆盖软删除状态? A:使用乐观锁(如version字段),每次更新前校验版本号,防止并发覆盖。

Q3:软删除后,关联的外键数据该怎么处理? A:分两种情况:1) 外键被软删除——建议保留外键值,但查询时自动过滤;2) 外键被物理删除——需在软删除前检查关联完整性。

Q4:SQLAlchemy 2.0是否改变了软删除的实现方式? A:核心逻辑不变,但2.0推荐使用select()函数而非query,建议使用with_loader_criteriaSessionEvents替代旧的Query拦截。

Q5:大量软删除数据会影响查询性能吗? A:是的,因为每次查询都需要额外过滤条件,建议:1) 对is_deleted建立索引;2) 定期将超过时间阈值的记录迁移到历史表。


💡 扩展推荐:如果你需要更复杂的软删除(如多级软删除、层级结构软删除),建议研究sqlalchemy-softdeleteflask-sqlalchemy-softdelete社区插件,它们提供了开箱即用的解决方案。

相关资源:GitHub上的klen/mixerjrderuiter/sqlalchemy-soft-delete-mixin项目提供了更完整的软删除工具集。

抱歉,评论功能暂时关闭!