Become a sponsor

Repository 层是数据访问层,封装所有数据库查询操作。项目通过 BaseRepository 基类提供通用 CRUD 能力,子类通常只需声明类名,无需编写任何代码。
src/modules/system/position/repository.pyfrom modules.system.position.models import Position
from core.base_repository import BaseRepository
# ============================================================
# 岗位仓库类
# ============================================================
class PositionRepository(BaseRepository[Position]):
"""岗位数据仓库,继承基础仓库获得通用 CRUD 能力"""
pass
# ============================================================
# 仓库实例
# ============================================================
position_repo = PositionRepository(Position)class PositionRepository(BaseRepository[Position]):BaseRepository[Position] 是泛型基类,传入模型类型后自动获得以下能力:
| 方法 | 说明 |
|---|---|
get_by_id(id) | 根据 ID 查询单条(自动过滤软删除) |
get_by_ids(ids) | 根据 ID 列表查询多条 |
get_one(**kwargs) | 按等值条件查询单条(跳过 None 值) |
get_all(**kwargs) | 按等值条件查询全部(自动过滤软删除) |
exists_by_field(field, value) | 判断字段值是否存在 |
count(**kwargs) | 按条件计数 |
filter(*criterion) | 原始 SQLAlchemy 条件查询(不自动过滤软删除) |
filter_by(**kwargs) | 等值条件查询(自动过滤软删除,跳过 None) |
create(**kwargs) | 创建记录 |
update(id, **kwargs) | 更新记录 |
paginate(page, limit) | 分页查询 |
batch_delete(ids) | 批量软删除 |
position_repo = PositionRepository(Position)每个模块在文件末尾创建一个模块级单例 xxx_repo,业务代码通过导入此实例访问数据库,避免重复创建 Repository 对象。
大多数简单 CRUD 模块(如 Position、Level、Category)的 Repository 只需 pass。当需要以下场景时,才在子类中添加自定义方法:
示例(UserRepository 中的自定义方法):
class UserRepository(BaseRepository[User]):
def get_by_username(self, username: str):
"""根据用户名查询用户(含角色信息)"""
return self.filter(
User.username == username,
User.is_delete == 0
).first()pass:继承基类即可获得完整 CRUD 能力xxx_repo = XxxRepository(XModel),供 Service 层调用filter() 不自动过滤软删除:需要手动追加 Model.is_delete == 0 条件filter_by() / get_one() / get_all() 自动过滤软删除:且跳过 None 值,推荐优先使用Repository 层继承 BaseRepository 获得通用 CRUD 能力,等值查询方法(get_one/get_all/filter_by)自动过滤软删除且跳过 None 值,filter() 为原始出口需手动追加软删除条件。Repository 层不做响应封装,由 Service 层处理。