后端 CRUD 的工程化抽象：增删改查也能写出架构感-北京尧图网络

后端 CRUD 的工程化抽象增删改查也能写出架构感一、CRUD 被低估的能力CRUD 是后端开发中最基础的操作。也正因为它基础很多人觉得它不值得讨论——不就是写几个 SQL 嘛。但这个态度本身就是问题。在后端实习中我发现一个 CRUD 模块写得好不好差别不在代码量上而在新增一个字段时需要改几个地方1 个还是 5 个当需要加缓存时是在 Controller 里硬编码还是通过装饰器当需要做软删除时是改所有 SQL 还是抽象出一层好的 CRUD 抽象让变更只发生在一个地方。这篇文章讨论如何把 CRUD 写得有架构感。二、CRUD 的抽象分层flowchart TD A[Controller 层] --|参数校验响应格式化| B[Service 层] B --|业务逻辑编排| C[Repository 层] C --|数据访问抽象| D[数据源] B --|横切关注点| E[AOP 层] E -- E1[日志记录] E -- E2[权限校验] E -- E3[缓存管理] E -- E4[事务管理]三、实现泛型 CRUD 基类from abc import ABC, abstractmethod from typing import TypeVar, Generic, Optional from dataclasses import dataclass from datetime import datetime from functools import wraps import hashlib import json T TypeVar(T) # 实体类型 class NotFoundError(Exception): 资源不存在异常 pass class DuplicateError(Exception): 资源重复异常 pass class BaseRepository(Generic[T], ABC): 泛型 CRUD Repository 基类提供标准的 CRUD 操作子类只需实现少数几个抽象方法。这种设计让所有实体的 Repository 保持一致的接口风格。工程优势 - 新增实体只需继承并实现 _table_name 等方法 - 通用逻辑如软删除在基类中一次实现 - 子类可以覆盖任意方法做定制 property abstractmethod def _table_name(self) - str: 子类必须提供表名 ... abstractmethod def _row_to_entity(self, row: dict) - T: 子类必须实现将数据库行转换为实体 ... abstractmethod def _entity_to_row(self, entity: T) - dict: 子类必须实现将实体转换为数据库行 ... abstractmethod def _execute(self, sql: str, params: list) - any: 数据库执行入口依赖注入 ... # 标准 CRUD 操作 def find_by_id(self, entity_id: int) - Optional[T]: 按 ID 查询 sql ( fSELECT * FROM {self._table_name} fWHERE id ? AND deleted_at IS NULL ) rows self._execute(sql, [entity_id]) if not rows: return None return self._row_to_entity(rows[0]) def find_by_id_or_raise(self, entity_id: int) - T: 按 ID 查询不存在时抛异常 entity self.find_by_id(entity_id) if entity is None: raise NotFoundError( f{self._table_name} with id{entity_id} not found ) return entity def find_all( self, filters: Optional[dict] None, order_by: Optional[str] None, limit: Optional[int] None, offset: int 0, ) - list[T]: 带条件的分页查询 Args: filters: 过滤条件如 {status: active, user_id: 123} order_by: 排序如 created_at DESC limit: 每页数量 offset: 偏移量 sql fSELECT * FROM {self._table_name} WHERE deleted_at IS NULL params [] # 动态构建 WHERE 条件使用参数化查询防止 SQL 注入 if filters: for key, value in filters.items(): sql f AND {key} ? params.append(value) if order_by: sql f ORDER BY {order_by} if limit is not None: sql f LIMIT ? OFFSET ? params.extend([limit, offset]) rows self._execute(sql, params) return [self._row_to_entity(row) for row in rows] def create(self, entity: T) - T: 创建实体 row self._entity_to_row(entity) # 设置创建时间 row[created_at] datetime.now() row[updated_at] datetime.now() columns , .join(row.keys()) placeholders , .join([?] * len(row)) sql ( fINSERT INTO {self._table_name} ({columns}) fVALUES ({placeholders}) ) self._execute(sql, list(row.values())) # 返回创建的实体实际应从数据库获取自增 ID 等 return entity def update(self, entity_id: int, updates: dict) - T: 部分更新实体只更新提供的字段不覆盖未提供的字段。 # 先检查实体是否存在 self.find_by_id_or_raise(entity_id) updates[updated_at] datetime.now() set_clause , .join(f{k} ? for k in updates) sql ( fUPDATE {self._table_name} fSET {set_clause} fWHERE id ? AND deleted_at IS NULL ) params list(updates.values()) [entity_id] self._execute(sql, params) return self.find_by_id(entity_id) def delete(self, entity_id: int, soft: bool True) - None: 删除实体 Args: soft: True软删除设置 deleted_atFalse物理删除 self.find_by_id_or_raise(entity_id) if soft: sql ( fUPDATE {self._table_name} fSET deleted_at ?, updated_at ? fWHERE id ? ) now datetime.now() self._execute(sql, [now, now, entity_id]) else: sql fDELETE FROM {self._table_name} WHERE id ? self._execute(sql, [entity_id]) def count(self, filters: Optional[dict] None) - int: 统计数量 sql fSELECT COUNT(*) as cnt FROM {self._table_name} WHERE deleted_at IS NULL params [] if filters: for key, value in filters.items(): sql f AND {key} ? params.append(value) result self._execute(sql, params) return result[0][cnt] if result else 0 # ---- 装饰器增强缓存层 ---- def cached(ttl: int 300): Repository 方法缓存装饰器用法在需要缓存的方法上添加 cached(ttl300) def decorator(func): wraps(func) def wrapper(self, *args, **kwargs): # 生成缓存键 cache_key ( f{self._table_name}:{func.__name__}: f{hashlib.md5(str((args, kwargs)).encode()).hexdigest()} ) # 尝试从缓存读取 cached_value self._cache_get(cache_key) if cached_value is not None: return cached_value # 缓存未命中执行原方法 result func(self, *args, **kwargs) self._cache_set(cache_key, result, ttl) return result return wrapper return decorator四、工程价值4.1 变更隔离当需要给所有实体加「软删除」时只要在基类的delete方法中实现一次所有子类自动获得该能力。这就是单一变更点的设计——修改只发生在一个地方。4.2 接口一致性所有子类的 CRUD 方法签名一致团队成员切换维护不同实体时不需要重新熟悉接口。4.3 测试简化基类的通用逻辑如 find_all 的动态查询构建只需测试一次。子类的测试可以专注于自定义逻辑。五、总结把 CRUD 写好不是把增删改查写得更快而是让变更与扩展更安全。抽象的价值不在当前——在一个月后需要加一个新字段、三个月后需要加缓存、半年后需要做软删除时——你会发现好的抽象让这些变更只影响一个地方而不需要在 50 个文件里逐行修改。

后端 CRUD 的工程化抽象：增删改查也能写出架构感

相关推荐

用 AI 重构刷题笔记：从零散记录到结构化知识库

编程Agent决策仲裁层（DAL）：轻量级序列决策建模实践

33 SFT、RLHF、DPO：模型对齐到底在对齐什么

更多资讯

从环境配置到训练曲线：DQN与DDQN在Atari上的完整工程实践

DQN源码实战：在Atari Breakout上复现训练与避坑指南

基于gym Atari的DQN实战：环境配置、算法原理与变体实现详解

AI实战入门：从命令行到本地部署的30天可交付路径

磁力链接转种子文件：3分钟永久保存稀缺资源教程

基于改进Unet的多模态MRI融合脑梗死分割实战

今日更新

Play Framework 迁移指南：移除 GlobalSettings，全面转向依赖注入（Scala 与 Java）

Numba 类型推断机制详解：从 Numba IR 到编译期类型重建的完整原理与实践

企业级项目dragonballz_e159-1的技术架构与实现方案

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

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