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

• 实现所有常见CRUD操作

• 支持分页、排序、存在性检查、计数

• 通过Python泛型保证类型安全

• 允许自定义实体与模型的映射

• 允许每个仓库自定义过滤条件

• 保持代码整洁、可扩展、易测试

from dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom uuid import UUID@dataclass(kw_only=True)class EntityBase:    id: UUID | None = None    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))    updated_at: datetime = field(default_factory=lambda: 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", bound=EntityBase)SqlAlchemyModel = TypeVar("SqlAlchemyModel", bound=Base)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_by=order_by, ordering=ordering)        )        # 分页        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)

当你创建具体仓库时：

class UserRepository(SqlAlchemyAbstractRepository[User, UserModel]):    ...

就相当于说：“我要一份User饭装在UserModel餐具里。”

IDE现在就能准确知道：

class SqlAlchemyUserRepository(    SqlAlchemyAbstractRepository[User, UserModel],):    model = UserModel    def _entity_to_model(self, entity: User) -> UserModel:        model = UserModel(            name=entity.name,            email=entity.email,            role=entity.role,        )        # 如果实体已有ID（更新场景），保持ID        if entity.id:            model.id = entity.id        return model    def _model_to_entity(self, model: UserModel) -> User:        return User(            id=model.id,            name=model.name,            email=model.email,            role=model.role,            created_at=model.created_at,            updated_at=model.updated_at,        )    def _get_filters(self, **filters):        """支持三种过滤条件：id、email、role"""        conditions = []        if"id_filter"in filters:            conditions.append(UserModel.id == filters["id_filter"])        if"email_filter"in filters:            conditions.append(UserModel.email == filters["email_filter"])        if"role_filter"in filters:            conditions.append(UserModel.role == filters["role_filter"])        return conditions

# 查询管理员admins = await user_repo.list_all(    role_filter="admin",    page=1,    limit=20)# 按邮箱查找单个用户user = await user_repo.get(email_filter="john@example.com")# 检查用户是否存在exists = await user_repo.exists(email_filter="john@example.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()            # 检查是否是邮箱重复            if"ix_users_email"in str(e):                raise UserAlreadyExistsError(entity.email)            raise

class SqlAlchemyUserRepository(...):    # ... 前面的代码    asyncdef get_by_email(self, email: str) -> User | None:        """按邮箱获取用户（业务常用）"""        returnawait self.get(email_filter=email)    asyncdef get_active_admins(self) -> List[User]:        """获取活跃管理员（业务特定）"""        returnawait self.list_all(            role_filter="admin",            status_filter="active"        )