Repository层规范

概述

Repository 层是数据访问层，封装所有数据库操作。各模块 Repository 继承 BaseRepository[Model]，获得通用 CRUD 能力，子类按需添加自定义查询。

设计原则

• Repository 只做数据访问原语，不处理 R 封装/序列化/业务逻辑
• 软删除由基类自动处理，子类无需关心
• 查询方法自动跳过 None 值条件

BaseRepository 继承

以岗位模块为例，src/modules/system/position/repository.py 完整内容：

"""modules/system/position 数据访问层。"""
from modules.system.position.models import Position
from core.base_repository import BaseRepository


class PositionRepository(BaseRepository[Position]):
    """岗位数据仓库，继承基础仓库获得通用 CRUD 能力"""
    pass


# 模块级单例
position_repo = PositionRepository(Position)

单例模式

每个模块在文件末尾创建 Repository 实例（如 position_repo），供 Service 层使用。避免每次请求创建新实例。基类构造函数接收模型类：PositionRepository(Position)。

查询方法选择表

方法	软删过滤	自动跳过 None	适用场景
get_by_id(id)	是	-	按 ID 查询单条
get_by_ids(ids)	是	-	按 ID 列表批量查询
get_one(**filters)	是	是	按等值条件查单条
get_all(**filters)	是	是	按等值条件查多条
count(**filters)	是	是	统计记录数
exists(obj_id)	是	-	检查记录是否存在
exists_by_field(field, value, exclude_id)	是	-	唯一性校验
filter_by(**kwargs)	是	是	等值查询，返回 Query
filter(*criterion)	否	-	原始查询，复杂条件
paginate(page, size, query)	是	-	分页查询

filter 与 filter_by 的区别

• filter_by(**kwargs)：等值查询，自动软删过滤，跳过 None 值，返回 Query
• filter(*criterion)：原始查询，不自动软删过滤，用于 in_/!=/and_/区间等复杂条件，需调用方自行追加 is_delete == 0

常用查询示例

等值查询

# 查询单条：自动软删过滤，跳过 None
user = user_repo.get_one(username='admin', status=1)

# 查询多条：自动软删过滤
positions = position_repo.get_all(status=1)

# 统计数量
count = position_repo.count(status=1)

# 唯一性校验
exists = position_repo.exists_by_field('name', '开发工程师', exclude_id=5)

复杂条件查询

# 使用 filter（需自行追加 is_delete 条件）
users = user_repo.filter(
 User.dept_id.in_([1, 2, 3]),
 User.is_delete == 0
).all()

# 区间查询
articles = article_repo.filter(
 Article.create_time >= start_date,
 Article.create_time <= end_date,
 Article.is_delete == 0
).all()

# 使用 filter_by + 排序
positions = position_repo.filter_by(status=1).order_by(
 Position.sort.asc()
).all()

分页查询

# 自定义 query 分页
query = position_repo.filter_by(status=1).order_by(Position.sort.asc())
items, total = position_repo.paginate(page_no=1, page_size=10, query=query)

写操作

创建

# 创建记录（不提交事务，由中间件统一管理）
position_repo.create(name='开发工程师', status=1, sort=1)

更新

# 按 ID 更新，返回受影响行数
count = position_repo.update(obj_id=1, data={'name': '高级开发工程师', 'sort': 2})

删除

# 软删除，返回实际删除条数
count = position_repo.batch_delete("1,2,3")

删除的 R 封装

Repository 的 batch_delete 只做数据原语，R 响应由 Service 层 BaseService.batch_delete_with_r(repo, ids) 统一产出。禁止在 Repository 层直接返回 R 对象。

自定义查询方法

子类可添加模块特有的查询方法：

class UserRepository(BaseRepository[User]):
    """用户数据仓库"""

    def get_by_username(self, username: str):
        """根据用户名查询"""
        return self.get_one(username=username)

    def get_users_by_dept(self, dept_id: int):
        """查询部门下所有用户"""
        return self.get_all(dept_id=dept_id)

    def get_users_with_role(self, role_id: int):
        """查询拥有指定角色的用户"""
        return self.filter(
            User.is_delete == 0
        ).join(
            UserRole, UserRole.user_id == User.id
        ).filter(
            UserRole.role_id == role_id
        ).all()

数据清洗

create 和 update 方法内置 _clean_model_data()，自动将 camelCase 键转为 snake_case，并仅保留模型真实列字段：

# 传入 camelCase 键，自动转换
position_repo.create(positionName='开发', status=1)
# 内部自动转为 position_name → 查找模型列 → 忽略不存在的字段

总结

Repository 层通过继承 BaseRepository[Model] 获得完整的 CRUD 能力。查询方法分两类：等值便捷方法（get_one/get_all/filter_by 等，自动软删过滤）和原始查询出口（filter，需自行处理软删）。写操作由 Repository 做数据原语，R 响应由 Service 层统一封装。模块级单例模式避免重复创建实例。