部门/职级/岗位

部门、职级、岗位是组织架构管理的三个核心模块。部门采用树形结构，职级和岗位为简单 CRUD 模块，均继承 BaseService。用户表通过 dept_id、position_id、level_id 三个外键关联这三个模块。

部门管理

部门模型

# src/modules/system/dept/models.py
# ============================================================
# 部门模型
# ============================================================
class Dept(BaseModel, BaseDB):
    """部门模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "dept"
    __table_comment__ = "部门表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 部门名称
    name = Column(String(150), nullable=False, index=True, comment="部门名称")
    # 部门类型：1-公司 2-子公司 3-部门 4-小组
    type = Column(Integer, default=0, server_default=text('0'), comment="部门类型：1-公司 2-子公司 3-部门 4-小组")
    # 上级部门ID
    parent_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="上级部门ID")
    # 部门排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="部门排序")
    # 备注
    note = Column(String(255), nullable=True, comment="备注")

    # ============================================================
    # 内置方法
    # ============================================================
    def __str__(self):
        """返回部门名称作为字符串表示"""
        return '部门{}'.format(self.name)

树形结构

部门通过 parent_id 字段形成树形结构，与菜单类似：

公司总部 (parent_id=0)
├── 技术部 (parent_id=1)
│ ├── 前端组 (parent_id=2)
│ └── 后端组 (parent_id=2)
└── 市场部 (parent_id=1)

删除前校验

部门删除时需检查是否存在下级部门或用户引用，避免组织架构悬空：

# src/modules/system/dept/service.py
from src.common.base_service import BaseService
from src.core.exceptions import BizError
from src.modules.system.dept.repository import DeptRepository
from src.modules.system.user.repository import UserRepository


class DeptService(BaseService):
    """部门服务类"""
    repository = DeptRepository
    form = DeptForm
    model_name = "部门"

    @classmethod
    def before_delete(cls, ids):
        """
        删除前校验：检查下级部门和用户引用

        Args:
            ids: 待删除的部门ID列表

        Raises:
            BizError: 存在下级部门或用户引用时抛出
        """
        for did in ids:
            # 检查是否存在下级部门
            if DeptRepository.count_children(did) > 0:
                raise BizError(f"部门ID {did} 存在下级部门，请先删除下级部门")
            # 检查是否存在用户引用
            if UserRepository.count_by({'dept_id': did}) > 0:
                raise BizError(f"部门ID {did} 已被用户引用，请先调整用户部门")

职级管理

职级模型

# src/modules/system/level/models.py
# ============================================================
# 职级模型
# ============================================================
class Level(BaseModel, BaseDB):
    """职级模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "level"
    __table_comment__ = "职级表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 职级名称
    name = Column(String(255), nullable=False, index=True, comment="职级名称")
    # 职级状态：1-在用 2-停用
    status = Column(Integer, default=0, server_default=text('0'), index=True, comment="职级状态：1-在用 2-停用")
    # 职级排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="职级排序")

    # ============================================================
    # 内置方法
    # ============================================================
    def __repr__(self):
        """返回职级名称作为字符串表示"""
        return '职级：{}'.format(self.name)

表单验证

# src/modules/system/level/schema.py
from wtforms import StringField, IntegerField
from wtforms.validators import DataRequired, Length, NumberRange, Optional
from src.common.forms.base_form import BaseForm
from src.modules.system.level.repository import LevelRepository


class LevelForm(BaseForm):
    """职级创建/编辑表单"""
    name = StringField('职级名称', validators=[
        DataRequired(message='职级名称不能为空'),
        Length(max=150, message='职级名称长度不得超过150个字符')
    ])
    status = IntegerField('职级状态', validators=[
        DataRequired(message='职级状态不能为空'),
        NumberRange(min=1, max=2, message='职级状态值在1~2之间')
    ])
    sort = IntegerField('职级排序', validators=[
        Optional(),
        NumberRange(min=0, max=99999, message='职级排序值在0~99999之间')
    ])

    def validate_name(self, field):
        """职级名称唯一性校验"""
        if not field.data:
            return
        current_id = self.id.data if self.id.data else None
        if LevelRepository.exists_by({'name': field.data}, exclude_id=current_id):
            raise ValidationError("职级名称已存在")


class LevelStatusForm(BaseForm):
    """职级状态更新表单"""
    id = IntegerField('职级ID', validators=[
        DataRequired(message='职级ID不能为空'),
        NumberRange(min=1, message='职级ID必须大于0')
    ])
    status = IntegerField('职级状态', validators=[
        DataRequired(message='职级状态不能为空'),
        NumberRange(min=1, max=2, message='职级状态值在1~2之间')
    ])

业务服务

职级模块继承 BaseService，声明差异点即可获得完整 CRUD 能力。删除前校验用户引用，支持 Excel 导入导出：

# src/modules/system/level/service.py
from src.common.base_service import BaseService
from src.core.exceptions import BizError
from src.modules.system.level.schema import LevelForm, LevelStatusForm
from src.modules.system.level.repository import LevelRepository
from src.modules.system.user.repository import UserRepository


class LevelService(BaseService):
    """职级服务类，继承 BaseService 获得标准 CRUD 能力"""
    repository = LevelRepository
    form = LevelForm
    status_form = LevelStatusForm
    model_name = "职级"

    @classmethod
    def before_delete(cls, ids):
        """删除前校验：存在用户引用该职级时禁止删除"""
        for lid in ids:
            if UserRepository.count_by({'level_id': lid}) > 0:
                raise BizError(f"职级ID {lid} 已被用户引用，无法删除")


# 模块级代理函数
get_level_page = LevelService.get_page
get_level_detail = LevelService.get_detail
add_level = LevelService.add
update_level = LevelService.update
delete_levels = LevelService.remove
update_level_status = LevelService.update_status
get_level_list = LevelService.get_list

岗位管理

岗位模型

# src/modules/system/position/models.py
# ============================================================
# 岗位模型
# ============================================================
class Position(BaseModel, BaseDB):
    """岗位模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "position"
    __table_comment__ = "岗位表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 岗位名称
    name = Column(String(255), nullable=False, index=True, comment="岗位名称")
    # 岗位状态：1-在用 2-停用
    status = Column(Integer, default=0, server_default=text('0'), index=True, comment="岗位状态：1-在用 2-停用")
    # 岗位排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="岗位排序")

    # ============================================================
    # 内置方法
    # ============================================================
    def __repr__(self):
        """返回岗位名称作为字符串表示"""
        return '岗位：{}'.format(self.name)

表单验证

# src/modules/system/position/schema.py
from wtforms import StringField, IntegerField
from wtforms.validators import DataRequired, Length, NumberRange, Optional
from src.common.forms.base_form import BaseForm
from src.modules.system.position.repository import PositionRepository


class PositionForm(BaseForm):
    """岗位创建/编辑表单"""
    name = StringField('岗位名称', validators=[
        DataRequired(message='岗位名称不能为空'),
        Length(max=150, message='岗位名称长度不得超过150个字符')
    ])
    status = IntegerField('岗位状态', validators=[
        DataRequired(message='岗位状态不能为空'),
        NumberRange(min=1, max=2, message='岗位状态值在1~2之间')
    ])
    sort = IntegerField('岗位排序', validators=[
        Optional(),
        NumberRange(min=0, max=99999, message='岗位排序值在0~99999之间')
    ])

    def validate_name(self, field):
        """岗位名称唯一性校验"""
        if not field.data:
            return
        current_id = self.id.data if self.id.data else None
        if PositionRepository.exists_by({'name': field.data}, exclude_id=current_id):
            raise ValidationError("岗位名称已存在")


class PositionStatusForm(BaseForm):
    """岗位状态更新表单"""
    id = IntegerField('岗位ID', validators=[
        DataRequired(message='岗位ID不能为空'),
        NumberRange(min=1, message='岗位ID必须大于0')
    ])
    status = IntegerField('岗位状态', validators=[
        DataRequired(message='岗位状态不能为空'),
        NumberRange(min=1, max=2, message='岗位状态值在1~2之间')
    ])

数据访问层

# 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):
    """岗位数据仓储，继承 BaseRepository 获得通用 CRUD 能力"""
    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',
        )

业务服务

# src/modules/system/position/service.py
from src.common.base_service import BaseService
from src.core.exceptions import BizError
from src.modules.system.position.schema import PositionForm, PositionStatusForm
from src.modules.system.position.repository import PositionRepository
from src.modules.system.user.repository import UserRepository


class PositionService(BaseService):
    """岗位服务类，继承 BaseService 获得标准 CRUD 能力"""
    repository = PositionRepository
    form = PositionForm
    status_form = PositionStatusForm
    model_name = "岗位"

    @classmethod
    def before_delete(cls, ids):
        """删除前校验：存在用户引用该岗位时禁止删除"""
        for pid in ids:
            if UserRepository.count_by({'position_id': pid}) > 0:
                raise BizError(f"岗位ID {pid} 已被用户引用，无法删除")


# 模块级代理函数
get_position_page = PositionService.get_page
get_position_detail = PositionService.get_detail
add_position = PositionService.add
update_position = PositionService.update
delete_positions = PositionService.remove
update_position_status = PositionService.update_status
get_position_list = PositionService.get_list

通用 CRUD 接口

职级和岗位模块继承 BaseService，自动生成标准接口：

接口	方法	权限节点	说明
/api/v1/level/page	GET	sys:level:page	职级分页列表
/api/v1/level/list	GET	sys:level:list	职级列表（无分页）
/api/v1/level/detail/{id}	GET	sys:level:detail	职级详情
/api/v1/level/add	POST	sys:level:add	新增职级
/api/v1/level/update	PUT	sys:level:update	编辑职级
/api/v1/level/delete/{id}	DELETE	sys:level:delete	删除职级
/api/v1/level/batchDelete	DELETE	sys:level:batchDelete	批量删除
/api/v1/level/status	PUT	sys:level:status	设置状态
/api/v1/level/import	POST	sys:level:import	导入 Excel
/api/v1/level/export	GET	sys:level:export	导出 Excel
/api/v1/position/page	GET	sys:position:page	岗位分页列表
/api/v1/position/list	GET	sys:position:list	岗位列表（无分页）
/api/v1/position/detail/{id}	GET	sys:position:detail	岗位详情
/api/v1/position/add	POST	sys:position:add	新增岗位
/api/v1/position/update	PUT	sys:position:update	编辑岗位
/api/v1/position/delete/{id}	DELETE	sys:position:delete	删除岗位
/api/v1/position/batchDelete	DELETE	sys:position:batchDelete	批量删除
/api/v1/position/status	PUT	sys:position:status	设置状态

Excel 导入导出

职级模块支持 Excel 导入导出（import_level / export_level），导入时逐行校验并返回成功/失败条数。岗位模块结构相同，可按需添加导入导出功能。

总结

组织架构模块具备以下特点：

1. 部门树形：parent_id 形成多级部门结构，删除前校验下级部门和用户引用
2. 职级/岗位：简单 CRUD，继承 BaseService，声明差异点即可
3. 表单验证：WTForms Field 约束（DataRequired/Length/NumberRange）
4. 数据仓库：BaseRepository 提供通用 CRUD 能力，模块级代理函数
5. 用户关联：用户表通过 dept_id、position_id、level_id 关联
6. 删除保护：before_delete 钩子防止引用悬空