这个 Python 泛型仓库让你少写 80% 重复代码（附代码）

本文约4000字建议阅读5分钟本文介绍了用 Python 泛型和 SQLAlchemy 实现通用仓库告别重复 CRUD。你还在为每个实体手写CRUD这个Python泛型仓库模式让你一次编写随处复用一个真实场景刚接手一个FastAPI项目打开代码库UserRepository、ProductRepository、OrderRepository……每个文件都在重复同样的save、get、update、delete逻辑。复制粘贴了8次之后我开始怀疑人生——我们真的需要为每个数据表写一遍相同的代码吗如果你也有同样的困惑今天这篇文章会给你一个答案。我将带你用Python泛型和SQLAlchemy实现一个类型安全、可扩展、可复用的通用仓库模式让你从此告别重复的CRUD代码。重复的代码重复的痛苦在大多数FastAPI或SQLAlchemy项目中仓库层Repository长这样class UserRepository: def __init__(self, session: AsyncSession): self._session session asyncdef save(self, user: User) - User: model UserModel(nameuser.name, emailuser.email) self._session.add(model) await self._session.flush() await self._session.refresh(model) return self._to_entity(model) asyncdef get(self, user_id: UUID) - User | None: result await self._session.scalar( select(UserModel).where(UserModel.id user_id) ) return self._to_entity(result) if result elseNone # ... 更多方法然后你创建ProductRepository——复制粘贴。OrderRepository——再次复制粘贴。每个仓库都包含相同的CRUD逻辑相同的分页逻辑相同的错误处理相同的SQLAlchemy操作模式唯一变化的只有三样东西实体类型如UserORM模型类型如UserModel实体与模型之间的映射⚠️ 注意这种重复代码是“复制粘贴综合症”的典型表现90%的团队在这里踩坑——当业务逻辑需要修改时你要在8个仓库里改8遍漏改一个就是Bug。解决方案一个通用的抽象仓库一个设计良好的通用仓库应该做到实现所有常见CRUD操作支持分页、排序、存在性检查、计数通过Python泛型保证类型安全允许自定义实体与模型的映射允许每个仓库自定义过滤条件保持代码整洁、可扩展、易测试下面是一份生产级的实现代码。核心组件实体基类首先需要一个所有领域实体共享的基类保证统一的结构from dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom uuid import UUIDdataclass(kw_onlyTrue)class EntityBase: id: UUID | None None created_at: datetime field(default_factorylambda: datetime.now(timezone.utc)) updated_at: datetime field(default_factorylambda: datetime.now(timezone.utc))辅助工具异常与排序class DatabaseException(Exception): 数据库操作异常的统一包装 passfrom enum import StrEnumclass Ordering(StrEnum): 排序方向类型安全 asc asc desc desc通用仓库实现这是整个模式的核心。我把它拆成两部分讲解但你可以直接复制使用。from abc import ABC, abstractmethodfrom typing import Any, Generic, List, TypeVarimport sqlalchemyfrom sqlalchemy import asc, desc, func, selectfrom sqlalchemy.exc import IntegrityError, SQLAlchemyErrorfrom sqlalchemy.ext.asyncio import AsyncSession# 假设你的Base类在这里定义from .... import Basefrom domain.value_objects.ordering import Orderingfrom domain.entities.base import EntityBasefrom domain.exceptions.common import DatabaseExceptionEntity TypeVar(Entity, boundEntityBase)SqlAlchemyModel TypeVar(SqlAlchemyModel, boundBase)class SqlAlchemyAbstractRepository(ABC, Generic[Entity, SqlAlchemyModel]): # 子类必须指定具体的ORM模型类 model: type[SqlAlchemyModel] def __init__(self, session: AsyncSession) - None: self._session session asyncdef save(self, entity: Entity) - Entity: 保存实体返回包含数据库生成字段如ID的完整实体 model self._entity_to_model(entity) self._session.add(model) await self._session.flush() await self._session.refresh(model) return self._model_to_entity(model) asyncdef update( self, fields_to_update: dict[str, Any], **filters, ) - int: 根据过滤条件更新字段返回受影响的行数 try: filter_conditions self._get_filters(**filters) query ( sqlalchemy.update(self.model) .where(*filter_conditions) .values(fields_to_update) ) result await self._session.execute(query) await self._session.flush() return result.rowcount # type: ignore[attr-defined] except IntegrityError as exception: await self._session.rollback() raise exception except SQLAlchemyError as exception: await self._session.rollback() raise DatabaseException from exception asyncdef list_all( self, page: int 1, limit: int 10, order_by: str created_at, ordering: Ordering Ordering.asc, **filters, ) - List[Entity]: 分页列表查询支持排序和过滤 query select(self.model) filter_conditions self._get_filters(**filters) query query.where(*filter_conditions) # 排序 query query.order_by( self._get_order_expression(order_byorder_by, orderingordering) ) # 分页 offset (page - 1) * limit query query.offset(offset).limit(limit) result await self._session.execute(query) models result.scalars().all() return [self._model_to_entity(model) for model in models] asyncdef get( self, **filters, ) - Entity | None: 根据过滤条件获取单个实体 query select(self.model) filter_conditions self._get_filters(**filters) query query.where(*filter_conditions) model await self._session.scalar(query) return self._model_to_entity(model) if model elseNone asyncdef exists( self, **filters, ) - bool: 检查是否存在满足条件的记录 query select(self.model) filter_conditions self._get_filters(**filters) query query.where(*filter_conditions) result await self._session.scalar(query) return result isnotNone asyncdef delete( self, **filters, ) - int: 根据过滤条件删除记录返回删除的行数 try: query sqlalchemy.delete(self.model) filter_conditions self._get_filters(**filters) query query.where(*filter_conditions) result await self._session.execute(query) await self._session.flush() return result.rowcount # type: ignore[attr-defined] except SQLAlchemyError as e: await self._session.rollback() raise DatabaseException from e asyncdef count( self, **filters, ) - int: 统计满足条件的记录数 filter_conditions self._get_filters(**filters) return ( await self._session.scalar( select(func.count()).select_from(self.model).where(*filter_conditions) ) or0 ) staticmethod abstractmethod def _model_to_entity(model: SqlAlchemyModel) - Entity: 将ORM模型转换为领域实体——子类必须实现 raise NotImplementedError(Subclasses must implement _model_to_entity) staticmethod abstractmethod def _entity_to_model(entity: Entity) - SqlAlchemyModel: 将领域实体转换为ORM模型——子类必须实现 raise NotImplementedError(Subclasses must implement _entity_to_model) abstractmethod def _get_filters(self, **filters) - List[Any]: 将业务层过滤条件转换为SQLAlchemy查询条件——子类可重写 return [] staticmethod def _get_order_expression( order_by: str, ordering: Ordering ) - sqlalchemy.UnaryExpression[str]: 生成排序表达式 if ordering Ordering.asc: return asc(order_by) return desc(order_by)泛型解析用生活化类比理解如果上面这段代码让你有点晕我用一个类比帮你理清泛型就像订餐平台的模板Entity TypeVar(Entity, boundEntityBase) —— 这就像“我要一份饭”但具体是盖浇饭还是炒饭后面再定Model TypeVar(Model, boundBase) —— 这就像“我要一个餐具”具体是碗还是盘子也后面再定SqlAlchemyAbstractRepository[Entity, Model] —— 这个组合就像“我要一份某种饭搭配某种餐具的套餐”当你创建具体仓库时class UserRepository(SqlAlchemyAbstractRepository[User, UserModel]): ...就相当于说“我要一份User饭装在UserModel餐具里。”IDE现在就能准确知道save() 接收User返回User_model_to_entity() 必须把UserModel映射成User过滤条件只接受对User有效的字段⚠️ 关键点Python虽然是动态语言但通过类型提示和泛型你可以获得编译时类型检查的能力。这在多人协作时能避免无数“不小心传错参数”的Bug。实战创建具体的UserRepository现在创建一个用户仓库你会发现只需要写三件事指定model类实现映射逻辑定义支持的过滤条件class SqlAlchemyUserRepository( SqlAlchemyAbstractRepository[User, UserModel],): model UserModel def _entity_to_model(self, entity: User) - UserModel: model UserModel( nameentity.name, emailentity.email, roleentity.role, ) # 如果实体已有ID更新场景保持ID if entity.id: model.id entity.id return model def _model_to_entity(self, model: UserModel) - User: return User( idmodel.id, namemodel.name, emailmodel.email, rolemodel.role, created_atmodel.created_at, updated_atmodel.updated_at, ) def _get_filters(self, **filters): 支持三种过滤条件id、email、role conditions [] ifid_filterin filters: conditions.append(UserModel.id filters[id_filter]) ifemail_filterin filters: conditions.append(UserModel.email filters[email_filter]) ifrole_filterin filters: conditions.append(UserModel.role filters[role_filter]) return conditions看到没 整个仓库就这么点代码。CRUD已经处理好了分页已经处理好了错误处理已经处理好了你的仓库只需要关注领域特有的逻辑。为什么_get_filters这么重要它让你的查询API既干净又灵活# 查询管理员admins await user_repo.list_all( role_filteradmin, page1, limit20)# 按邮箱查找单个用户user await user_repo.get(email_filterjohnexample.com)# 检查用户是否存在exists await user_repo.exists(email_filterjohnexample.com)不需要为每个查询写单独的SQL所有过滤条件统一通过_get_filters转换为查询条件。自定义错误处理保留灵活扩展的空间需要处理特定业务的数据库错误只需覆盖方法class SqlAlchemyUserRepository(...): # ... 前面的代码 asyncdef save(self, entity: User) - User: try: returnawait super().save(entity) except IntegrityError as e: await self._session.rollback() # 检查是否是邮箱重复 ifix_users_emailin str(e): raise UserAlreadyExistsError(entity.email) raise⚠️ 注意这里的关键是await self._session.rollback()——忘记回滚会让session处于异常状态后续操作都会失败。这是90%的人踩过的坑。添加自定义方法通用 ≠ 不能定制通用仓库不代表不能添加特定查询class SqlAlchemyUserRepository(...): # ... 前面的代码 asyncdef get_by_email(self, email: str) - User | None: 按邮箱获取用户业务常用 returnawait self.get(email_filteremail) asyncdef get_active_admins(self) - List[User]: 获取活跃管理员业务特定 returnawait self.list_all( role_filteradmin, status_filteractive )通用 ≠ 限制而是从强大的基础上开始。真实项目效果对比在重构一个中等规模的FastAPI项目后数据是这样的维度重构前重构后仓库数量8个8个单个仓库代码量250-400行30-50行CRUD重复代码每个仓库重复0全部复用修改分页逻辑改8个地方改1个地方类型安全❌ 随意传参✅ 编译时检查核心洞察这种模式不仅减少了代码量更重要的是——逻辑集中在一处修改一次生效全局Bug率显著下降。为什么这个模式值得你采用1. DRY原则落地写一次修一次处处生效。2. 一致性保障所有仓库行为统一新人上手零学习成本。3. 类型安全告别Any和随意传递的字典IDE能给你准确的代码补全。4. 可测试性测试一次基类所有仓库都得到测试覆盖。5. 可维护性想加软删除在基类改一次所有仓库自动支持。6. 灵活性需要特殊行为覆盖方法即可基类不限制你。写在最后从复制粘贴8个仓库到用泛型基类一行行抽象出来这个过程让我意识到一件事好的抽象不是炫技而是当你需要修改代码时发现只需要改一个地方。通用仓库模式在Python生态中并不算新但它结合async、SQLAlchemy和泛型后能给你的代码质量带来质的飞跃。下次你再新建一个实体时不用再写那300行CRUD只需30行映射和过滤逻辑。如果你正在维护一个数据访问层臃肿的项目建议逐个仓库迁移而不是一次性全量替换。先迁移一个非核心的仓库验证无误后再逐步推进。核心内容原理泛型抽象基类让CRUD逻辑一次性实现类型安全有保障实践子类只需实现映射和过滤所有操作自动获得避坑记得处理事务回滚自定义过滤用_get_filters统一入口编辑于腾凯校对孙英杰关于我们数据派THU作为数据科学类公众号背靠清华大学大数据研究中心分享前沿数据科学与大数据技术创新研究动态、持续传播数据科学知识努力建设数据人才聚集平台、打造中国大数据最强集团军。新浪微博数据派THU微信视频号数据派THU今日头条数据派THU