FastAPI实现软删除：从字段设计到查询过滤的完整实践-北京尧图网络

刚入行做后端的时候我写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(timezoneTrue), nullableTrue, defaultNone ) property def is_deleted(self) - bool: return self.deleted_at is not None这样定义的好处是所有模型都有统一的字段名和类型后面写通用查询过滤器、通用序列化逻辑都方便。时间统一用带时区的DateTime类型避免不同时区环境下的时间错乱。2.3 为什么deleted_at要用可空类型而不是defaultFalse这里有个最容易踩的点。如果列是可空的那么查询未删除数据的条件就是WHERE deleted_at IS NULLSQL用得上索引如果你用is_deleted Boolean NOT NULL DEFAULT FALSE条件就是WHERE is_deleted FALSE写法上也没问题但如果你在同一个表里既想高效查“未删除”、又想高效查“已删除”两个索引策略就需要额外设计。还有一个更实际的原因SQLAlchemy的defaultNone让新建记录的deleted_at直接就是NULL不会触发额外的默认值逻辑而如果你给Boolean列设置server_default迁移到生产库后历史数据的布尔值可能与新写入的不一致需要额外修数。我在项目里碰过一次“幽灵数据”问题就是Boolean字段在ORM层和数据库层默认值不一致造成的。生产库里老数据是NULL新数据是False查询条件写is_deleted False时NULL记录直接被过滤掉了几万条历史数据就这样突然“消失”了。换成deleted_at可空时间戳之后此类问题再没出现过。3. FastAPI里软删除的完整实操链路3.1 写一个通用的软删除ServiceFastAPI本身不限制你用什么架构但为了复用软删除逻辑我习惯封装一个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_codestatus.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_atdatetime.now(timezone.utc)) ) result await self.db.execute(stmt) await self.db.commit() if result.rowcount 0: raise HTTPException( status_codestatus.HTTP_404_NOT_FOUND, detail数据不存在或已删除, )这个BaseService把删除操作统一做了两层保护一是只更新未被删除的行二是通过rowcount判断操作是否真的影响了数据。既避免重复删除也避免对不存在的数据做无意义更新。3.2 FastAPI路由层的接入路由层只需要依赖注入这个Servicefrom 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(modelUser, dbdb) 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, retvalTrue) 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, retvalTrue) 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置回NULLasync def restore(self, obj_id: int): stmt ( update(self.model) .where(self.model.id obj_id) .values(deleted_atNone) ) 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删除了账号他的邮箱是aexample.com此时因为软删除只是把deleted_at置为非空所以那个邮箱在表里仍然占着唯一约束的位置。新用户注册也想用aexample.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。这个习惯在走查审计的时候价值极大。第三条上线前写一个删除-恢复-物理删除的完整测试用例。测试用例至少覆盖以下场景普通查询不返回已删除数据、已删除数据能通过管理接口查询、恢复后数据重新可见、重复删除不报错、唯一字段可重复使用、物理删除后数据彻底消失。这套测试跑通之后软删除在业务层就基本稳了。软删除不是复杂的架构但它就像屋檐的排水沟——平时没人注意真到暴雨天能不能把水流掉就看出设计水平了。希望这篇基于实际踩坑经验整理的内容能帮你把排水沟提前修好。

FastAPI实现软删除：从字段设计到查询过滤的完整实践

相关推荐

豆瓣影评情感分析：朴素贝叶斯文本分类实战

操作系统T1考题解析：进程状态迁移与上下文切换双线推演

Java中extends与implements的区别：从语法到设计选型

更多资讯

计算机网络实验全链路：从Boson Netsim拓扑到SOCKET编程实战

基于Python的微博舆情分析可视化系统：爬虫、情感分析与展示全流程

近红外光谱回归建模：物理约束驱动的深度学习方法

开题报告写作新思路：千笔与灵感风暴AI的分工协作指南

网红罗盘时钟：物理+代码+光影的时间装置入门指南

Advanced Installer 20.7.1：打造干净可复现的MSI安装包与升级链路

今日更新

Susi AI API实战指南：susi_alexa_skill如何用Node.js调用chat.json获取智能回答

换新电脑了？KeyStats 恢复码数据找回完全指南，端到端加密统计一键重建

重建二叉树：由前序与中序遍历序列还原二叉树的原理与 Java 实现（YCBlogs 剑指 Offer 精讲）

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？