Repository 层规范

概述

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

设计原则

• Repository 只做数据访问原语，不处理 R 封装/序列化/业务逻辑
• 所有方法均为 @classmethod，无需创建实例
• 软删除由基类自动处理，子类无需关心
• 查询方法自动跳过空值条件

标准模块结构

# src/modules/system/position/repository.py
from src.common.base_repository import BaseRepository
from src.common.constants.common import PAGE_SIZE
from src.modules.system.position.models import Position


class PositionRepository(BaseRepository):
    """岗位数据仓储"""
    model = Position

    @classmethod
    def page(cls, params: dict):
        """岗位分页查询"""
        return cls.find_page(
            page=int(params.get('pageNo', 1)),
            page_size=int(params.get('pageSize', PAGE_SIZE)),
            filters={'status': params.get('status')},
            like_filters={'name': params.get('name')},
            order_by='sort',
        )

    @classmethod
    def list_enabled(cls, params=None):
        """查询启用的岗位列表"""
        params = params or {}
        return cls.find_list(
            filters={'status': 1},
            like_filters={'name': params.get('name')},
            order_by='sort',
        )

必须实现的方法

方法	说明	何时需要
page(params)	分页查询	所有模块（Service 的 get_page 依赖此方法）
list_enabled(params)	无分页列表查询	需要下拉选择的模块（Service 的 get_list 依赖此方法）

BaseRepository 通用方法

方法	说明	自动过滤软删除
get_by_id(obj_id)	按主键查询单条	是
get_by_ids(ids)	按主键列表批量查询	是
first_by(filters)	按等值条件查询单条	是
exists_by(filters, exclude_id)	检查是否存在（唯一性校验）	是
count_by(filters)	按等值条件统计记录数	是
exists(obj_id)	检查记录是否存在	是
count_children(parent_id)	统计子记录数	是
find_page(...)	统一分页查询	是
find_list(...)	统一不分页查询	是
create(**kwargs)	创建并保存新实例	-
update_by_id(obj_id, data)	按主键更新字段	-
update_status(obj_id, status)	单独更新状态字段	-
delete_by_id(obj_id)	软删除单条记录	-
batch_delete(ids)	批量软删除	-
batch_restore(ids)	批量恢复已删除记录	-

使用示例

# 唯一性校验
if PositionRepository.exists_by({'name': '高级工程师'}, exclude_id=5):
    raise ValidationError('岗位名称已存在')

# 按 ID 查询
position = PositionRepository.get_by_id(1)

# 条件查询
user = UserRepository.first_by({'username': 'admin'})

# 统计引用
count = UserRepository.count_by({'position_id': 1})

# 分页查询
items, total, pages = PositionRepository.find_page(
    page=1, page_size=10,
    filters={'status': 1},
    like_filters={'name': '工程师'},
    order_by='sort',
)

开发要点

1. 所有方法均为 classmethod：无需创建实例，直接通过类调用
2. 必须实现 page(params)：Service 的 get_page 依赖此方法
3. 按需实现 list_enabled(params)：需要下拉选择的模块实现此方法
4. find_page / find_list 自动过滤软删除：且跳过空值查询条件
5. exists_by(filters, exclude_id) 用于唯一性校验：支持排除自身 ID
6. Repository 层不做响应封装：返回 SQLAlchemy 查询结果

总结

Repository 层继承 BaseRepository 获得通用 CRUD 能力，所有方法均为 classmethod。子类只需声明 model 并实现 page() / list_enabled() 方法。find_page / find_list 自动过滤软删除且跳过空值条件，exists_by 用于唯一性校验。