☰
FastAPI实现软删除:从字段设计到查询过滤的完整实践
2026/10/12 13:48:54 网站建设 项目流程

刚入行做后端的时候,我写DELETE端点那叫一个痛快,一句话await db.delete(obj)直接把数据从表里抹掉。直到有次在生成环境误删了一张业务表的关键记录,客户那边又拿不出备份,从那以后我给项目做删除功能就再也没敢“硬删过”。在FastAPI项目里做软删除,听起来就是个字段的事,但真要做得稳、查得对、不踩坑,牵扯到的设计点比很多人想象中要多得多。

这篇文章不聊理论,直接围绕FastAPI + SQLAlchemy(同步或异步都可以用)讲软删除的完整实现路径,包括为什么要用、字段怎么设计、service层怎么写、查询怎么自动过滤、唯一索引冲突怎么解、关联数据怎么处理。面向正在写接口的开发者,也适合想把删除逻辑做得更规范的技术负责人参考。

1. 从一次误删事故说起:为什么选择软删除

软删除的基本逻辑很简单:不是真的从数据库里把一行数据删掉,而是通过一个标记字段把它“隐藏”起来,正常业务查询看不到它,但数据本身还在表里躺着。等哪天真要销毁,再做物理删除。

听起来好像是绕了一圈,但实际业务里这个“绕一圈”能救很多人。

第一类场景是误操作恢复。没有任何系统能保证操作永远不出错,尤其是管理后台、批量操作、定时任务这种低关注度入口,一个筛选条件写错,影响的就是几百条数据。硬删除之后,恢复基本靠备份;而用了软删除,一条UPDATE就能把数据捞回来,恢复成本低到可以忽略。

第二类场景是审计和合规。现在很多业务都要留痕,尤其是涉及用户资产、订单、往来记录这类数据。用户主动注销账号、清理历史记录,如果直接硬删除,那审计链就断了,监管来查的时候你拿不出东西;软删除虽然不能让数据“看不见”,但至少能让业务层面的“已删除”和数据层面的“可追溯”共存。

第三类场景是关联数据保护。一本书删了,它的章节、评论、阅读记录如果跟着物理删掉,那这些历史数据的统计、分析、报表都会受影响。软删除后,主数据隐藏了,但子表数据仍然能保留下来,后续要做“回收站”甚至“延长保留期”也都有回旋余地。

FastAPI项目做软删除还有一个额外优势:FastAPI的依赖注入和路由结构很适合把软删除逻辑统一封装。你可以在BaseService里把“过滤未删除数据”这个事做成默认行为,所有资源接口自动继承,不用在每个路由里重复写WHERE deleted = 0,这也是我后来坚持在FastAPI项目里统一做软删除管理的原因之一。

当然,软删除也不是银弹。它会让唯一约束变得复杂,会让查询多一个过滤条件,还会让聚合统计口径需要额外定义。这些坑我在第4部分会逐个展开。但总体权衡,尤其对于ToB业务、SaaS系统、内容平台,软删除的收益远大于成本。

2. 核心设计思路:软删除字段该怎么建模

2.1 只用is_deleted够不够

新手最爱的方式是只加一个布尔字段is_deleted,默认False。这个方案最直接,但有几个问题。

首先,布尔字段没有时间信息。员工列表里删掉一个人,你只知道他被删了,不知道是什么时候删的、是谁删的。排查问题的时候,只看is_deleted,根本还原不了现场。

其次,布尔字段没法做“延迟删除”逻辑。比如一个业务规则要求数据删除后保留30天再物理清除,那你就必须记录deleted_at时间。

所以我一般建议用deleted_at可空时间戳做主标记。为空表示未删除,非空表示已删除,且删除时间自带了。这个设计在查询里也很干净:where deleted_at.is_(None)。如果你确实需要布尔字段方便日常索引或ORM映射,也可以两个字段都留,前端展示用is_deleted,后端逻辑判断统一用deleted_at,但要维护两个字段的同步。

2.2 用SQLAlchemy定义基础模型

在FastAPI项目里,我习惯定义一个公共的BaseModel,把软删除字段放进去,所有业务表继承它。

from datetime import datetime from sqlalchemy import DateTime, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class SoftDeleteMixin: deleted_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, default=None ) @property def is_deleted(self) -> bool: return self.deleted_at is not None

这样定义的好处是所有模型都有统一的字段名和类型,后面写通用查询过滤器、通用序列化逻辑都方便。时间统一用带时区的DateTime类型,避免不同时区环境下的时间错乱。

2.3 为什么deleted_at要用可空类型而不是default=False

这里有个最容易踩的点。如果列是可空的,那么查询未删除数据的条件就是WHERE deleted_at IS NULL,SQL用得上索引;如果你用is_deleted Boolean NOT NULL DEFAULT FALSE,条件就是WHERE is_deleted = FALSE,写法上也没问题,但如果你在同一个表里既想高效查“未删除”、又想高效查“已删除”,两个索引策略就需要额外设计。

还有一个更实际的原因:SQLAlchemy的default=None让新建记录的deleted_at直接就是NULL,不会触发额外的默认值逻辑;而如果你给Boolean列设置server_default,迁移到生产库后,历史数据的布尔值可能与新写入的不一致,需要额外修数。

我在项目里碰过一次“幽灵数据”问题,就是Boolean字段在ORM层和数据库层默认值不一致造成的。生产库里老数据是NULL,新数据是False,查询条件写is_deleted == False时,NULL记录直接被过滤掉了,几万条历史数据就这样突然“消失”了。换成deleted_at可空时间戳之后,此类问题再没出现过。

3. FastAPI里软删除的完整实操链路

3.1 写一个通用的软删除Service

FastAPI本身不限制你用什么架构,但为了复用软删除逻辑,我习惯封装一个BaseService。下面以SQLAlchemy 2.0 + 异步为例:

from fastapi import HTTPException, status from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession class BaseService: model = None def __init__(self, db: AsyncSession): self.db = db async def get(self, obj_id: int): stmt = select(self.model).where( self.model.id == obj_id, self.model.deleted_at.is_(None), ) result = await self.db.execute(stmt) obj = result.scalar_one_or_none() if obj is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="数据不存在", ) return obj async def soft_delete(self, obj_id: int): stmt = ( update(self.model) .where( self.model.id == obj_id, self.model.deleted_at.is_(None), ) .values(deleted_at=datetime.now(timezone.utc)) ) result = await self.db.execute(stmt) await self.db.commit() if result.rowcount == 0: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="数据不存在或已删除", )

这个BaseService把删除操作统一做了两层保护:一是只更新未被删除的行,二是通过rowcount判断操作是否真的影响了数据。既避免重复删除,也避免对不存在的数据做无意义更新。

3.2 FastAPI路由层的接入

路由层只需要依赖注入这个Service:

from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db from app.services.base import BaseService from app.models.user import User router = APIRouter(prefix="/users") def get_user_service(db: AsyncSession = Depends(get_db)): return BaseService(model=User, db=db) @router.delete("/{user_id}") async def delete_user( user_id: int, service: BaseService = Depends(get_user_service), ): await service.soft_delete(user_id) return {"message": "删除成功"}

注意一个细节,HTTP的DELETE语义其实是幂等性的,同一个ID删两次,第一次返回成功,第二次也应返回成功,但很多业务的习惯是重复删除要报错。我的建议是:对外接口保持幂等,返回成功即可;如果想审计,可以在日志或操作记录表里把第二次删除操作记为“无效操作”。不要因为重复删除给前端抛异常,用户体验很差。

3.3 自动过滤已删除数据:查询范围与全局过滤器

比“删除接口”更重要的是“查询接口绝不能把已删除数据带出来”。最稳妥的方式是做全局过滤。SQLAlchemy 2.0里有orm.with_loader_criteria或者重写Query,但最简单的方案是定义事件监听器。

from sqlalchemy import event from sqlalchemy.orm import Query @event.listens_for(Query, "before_compile", retval=True) def _add_soft_delete_filter(query: Query): for desc in query.column_descriptions: entity = desc["entity"] if entity is None: continue if hasattr(entity, "deleted_at"): query = query.filter(entity.deleted_at.is_(None)) break return query

事件监听器监听所有Query编译前的阶段,自动给带deleted_at字段的实体加上“未删除”过滤条件。看着挺省事,但我必须提醒你:这招对简单查询有效,但对复杂查询、联合查询、子查询,容易误伤或漏加。举两个例子:

  • 一个查询同时查询User和Order两个实体,如果两个实体都有deleted_at,事件监听器第一次break可能只给其中一个加了过滤,另一个还是全量。
  • 使用aliased、with_for_update、union这类高级查询,事件监听器很可能解析不到实体,过滤条件就丢了。

所以我的建议是:全局监听器适合做兜底,但每个业务查询的显式过滤条件不能省。尤其是涉及统计报表的接口,必须显式写deleted_at.is_(None),因为你无法保证监听器是否被正确触发。

3.4 让已删除数据能在特定场景下被查出来

软删除之后,管理后台总需要看“已删除列表”,也需要恢复、彻底删除的能力。这时我们不能再让全局过滤器拦截。我的做法是设计一个覆盖机制。

思路是用查询上下文标记来绕过过滤器:

from contextlib import contextmanager from sqlalchemy.orm import Session @contextmanager def include_deleted(session: Session | AsyncSession): previous = getattr(session.info, "include_deleted", False) session.info["include_deleted"] = True try: yield finally: session.info["include_deleted"] = previous

然后在全局过滤事件的回调里加上判定:

@event.listens_for(Query, "before_compile", retval=True) def _add_soft_delete_filter(query: Query): session = query.session if session is not None and session.info.get("include_deleted"): return query # ...原有的过滤逻辑

正常情况下业务查询走全局过滤;管理后台需要用include_deleted(session)上下文时,全局过滤就停了,可以查到包括已删除在内的全部数据。

这个上下文管理器一个很小,但它值得单独说,因为它决定了你的软删除方案到底能不能支持快速的“恢复/物理删除”功能。没有这层,恢复数据就只能去数据库手动刷字段,那就太落后了。

3.5 恢复与物理删除的API设计

在已删除列表里恢复记录,逻辑就是对合法数据把deleted_at置回NULL:

async def restore(self, obj_id: int): stmt = ( update(self.model) .where(self.model.id == obj_id) .values(deleted_at=None) ) result = await self.db.execute(stmt) await self.db.commit() if result.rowcount == 0: raise HTTPException(404, "记录不存在")

物理删除我建议单独设置权限,比如仅限超级管理员,并加上二次确认字段:

@router.delete("/{user_id}/hard", dependencies=[Depends(require_admin)]) async def hard_delete_user( user_id: int, confirm: bool = Query(...), service: BaseService = Depends(get_user_service), ): if not confirm: raise HTTPException(400, "缺少确认参数") await service.hard_delete(user_id) return {"message": "已物理删除"}

物理删除的接口默认不开放,即使开放了也要通过confirm参数强制前端二次确认。别嫌麻烦,这个接口一旦误触发,就回到文章开头的惨案现场了。

4. 常见问题与排查技巧实录

4.1 唯一索引冲突:软删除最大的坑

这是软删除方案里最典型、也最容易踩哭新手的坑。举个具体场景:用户表需要保证邮箱唯一,你给邮箱加了unique约束。用户A删除了账号,他的邮箱是a@example.com,此时因为软删除只是把deleted_at置为非空,所以那个邮箱在表里仍然占着唯一约束的位置。新用户注册,也想用a@example.com,就会直接报重复键错误。

解决办法大概有三种:

方案一:把唯一约束改成联合唯一约束,把deleted_at加进去。例如UNIQUE(email, deleted_at)。这样,正常记录里邮箱唯一,已删除的记录因为deleted_at时间戳不同,也可以保留同一条邮箱。听起来很完美,但MySQL里NULL值在多列唯一约束下不参与唯一校验,也就是说多条deleted_at为NULL的email重复记录会被数据库允许,约束失效。所以这个方案在MySQL下需要额外用COALESCE生成一个“影子字段”做约束,等于要加一列冗余。

方案二:删除时把唯一键改写掉。比如删除用户时把邮箱改成deleted_{user_id}_{email},这样就不会和新注册的邮箱冲突。缺点是你得想清楚所有用到邮箱的旧数据会不会因为邮箱被改而出问题,比如历史订单里存的外键是用户ID还好,如果业务表把邮箱作为关联键,那改动会牵一发动全身。

方案三:从设计上规避,不把“注销用户”的邮箱冲突视为问题。对于某些产品,用户注销后是否允许他人立刻注册同名邮箱,本身就是产品策略问题。我项目里的做法是:注册时检查的用户范围不只看deleted_at,还要加“是否在保留期”。规则是“删除后90天内不允许别人用这个邮箱注册”,既保证了唯一约束不会冲突,又遵守了合规保留期。

没有绝对正确的方案,关键是提前想清楚你的业务属主是谁。如果是SaaS,多租户环境下唯一约束本来就是租户ID + 业务字段的联合唯一,软删除带来的问题会小很多;如果是C端产品,邮箱手机号这类全局唯一约束,必须在设计阶段就想好冲突策略。

4.2 关联数据的孤儿与假外键

软删除还会让原本简单的外键关系变复杂。业务A有一个外键指向User,当User软删除后,业务A的记录仍然存在,但通过外键关联出来的用户信息,要么是空,要么是已删除的默认信息。

最常见的处理方式是关联查询时也过滤掉,表现成“用户不存在”,但这就可能导致业务A失去一条看似正常的数据。比如一个订单关联了一个被删除的用户,管理后台统计订单数量时,这个订单还在,但进到订单详情,用户信息却查不出来。

我现在的项目里对这种情况有两个策略:

一是对敏感关联字段,不直接删用户的主数据,而是把用户的name、email这些显示字段做脱敏处理。比如把邮箱改成带删除标记的***格式,这样即使关联关系变了,也能在报表里看到“这是一个已注销用户”。

二是在关联查询接口里同时返回deleted_at字段,前端根据该字段显示“该用户已注销”,而不是让接口直接404。从用户角度看,历史订单能看到但下单人显示“已注销”,比订单凭空消失更容易接受。

4.3 复杂查询中的软删除过滤

全局过滤器能覆盖大部分简单查询,但join、子查询、聚合函数这些复杂场景,很容易漏。我举一个实际遇到过的例子:统计每天新增用户数量,用func.date(created_at)分组统计,看起来简单,但没有显式过滤被软删除的用户。某天运营做了大规模注销,统计口径就乱了。

我的经验是:写复杂查询时,不要依赖全局过滤器,直接在语句里手动加条件。

async def count_active_users(db: AsyncSession) -> int: stmt = ( select(func.count()) .select_from(User) .where(User.deleted_at.is_(None)) ) result = await db.execute(stmt) return result.scalar()

尤其是报表和后台列表,宁可条件写重复,也不要漏。审计类的查询甚至应该反过来,明确查询已删除的数据,再去对比分析。

还有种情况是共用同一个子查询片段。建议把“未删除”条件做成一个公共方法:

def not_deleted(cls): return cls.deleted_at.is_(None)

这样每个查询直接写where(User.id.in_(subquery), not_deleted(User)),清晰且不容易漏,代码审查时一眼就能看出来。

4.4 异步session下的注意事项

FastAPI现在基本都用异步SQLAlchemy,软删除在异步环境里要注意三个点。

一是查询结果的类型。用scalars()取出来的多是model对象,如果你直接在对象上赋值obj.deleted_at = datetime.now(timezone.utc),就一定要保证该对象处于session的persistent状态,否则只改内存,数据库不会持久化。这点用同步ORM出问题不明显,因为同步session通常生命周期持续到函数结束,而异步session很容易在await之间被关闭或过期。

二是不要在事务中途做过多的await操作。一个软删除接口里如果既有delete又有后续关联数据的update,务必要显式管理事务边界,最好使用async with db.begin():,避免一个请求里出现多个隐式事务,导致部分成功部分失败。

三是时间函数建议统一使用datetime.now(timezone.utc)。别用datetime.utcnow(),后者在高版本Python里已经标记deprecated了。也别用数据库层的func.now()作为默认值之外的手动赋值。统一使用应用层UTC时间,跨时区部署、数据迁移时才不会乱。

4.5 已经被软删除的数据如何做统计

统计是软删除的另一个重灾区。如果你的产品指标是“注册用户数”,到底该不该去掉已软删除的用户?答案取决于指标定义,但代码层面必须支持两种口径。

我建议封装两个不同的repository方法,一个查含已删除总数,一个是只查未删除总数;在接口的docstring里写清楚口径,避免后面接手的同事看代码时猜。

class UserStatsService(BaseService): async def total_include_deleted(self) -> int: stmt = select(func.count()).select_from(self.model) result = await self.db.execute(stmt) return result.scalar() async def total_active_only(self) -> int: stmt = ( select(func.count()) .select_from(self.model) .where(self.model.deleted_at.is_(None)) ) result = await self.db.execute(stmt) return result.scalar()

别看这两个方法简单,命名越直白,后面越省心。我接手过一个项目,统计方法名里写的是count_users,里面既没用软删除过滤,也没有注释,结果运营用了半年错误数据。

5. 个人踩坑心得与几个使用建议

软删除这个方案,看着简单,但真落地到生产里,最考验人的不是代码怎么写,而是边界情况有没有想全面。

我第一次在FastAPI项目里做软删除,只加了deleted_at字段,所有查询都手动加过滤条件,但忘了唯一约束的坑。上线第二周,就有用户反馈注销后邮箱不能重新注册。当时我在排查日志里发现大量数据库唯一约束报错,才反应过来是软删除在索引层面留下了“尸体”。后来补了一个迁移脚本,把已删除用户的邮箱统一改掉,才恢复了注册功能。

还有一次是全局过滤器和分页查询联动出了问题。当时我用事件监听器自动过滤已删除数据,Post查询列表没问题,但某个聚合接口因为子查询里带limit,监听器没在子查询上生效,结果父查询关联出了一个被软删除的Post,直接把详情页搞崩了。从那以后我就立了一条规矩:全局过滤器只是兜底,业务查询必须显式写过滤条件,谁不写谁负责。

关于软删除,我还有三条个人建议想分享。

第一条,给软删除字段建索引。deleted_at虽然没有直接参与排序,但几乎所有查询都会用WHERE deleted_at IS NULL,如果表数据量大、查询频繁,这个条件的索引收益非常明显。建议联合索引字段根据查询习惯来建,比如经常按user_id过滤,考虑(user_id, deleted_at)联合索引。

第二条,操作记录要单独建表。软删除只是标记,真正要追溯“谁在什么时候删了什么东西”,操作日志得单独记录,甚至把操作前后的数据快照存成JSON。这个习惯在走查审计的时候价值极大。

第三条,上线前写一个删除-恢复-物理删除的完整测试用例。测试用例至少覆盖以下场景:普通查询不返回已删除数据、已删除数据能通过管理接口查询、恢复后数据重新可见、重复删除不报错、唯一字段可重复使用、物理删除后数据彻底消失。这套测试跑通之后,软删除在业务层就基本稳了。

软删除不是复杂的架构,但它就像屋檐的排水沟——平时没人注意,真到暴雨天,能不能把水流掉,就看出设计水平了。希望这篇基于实际踩坑经验整理的内容,能帮你把排水沟提前修好。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询