后端 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 个文件里逐行修改。