:::tip 部门/职级/岗位 部门、职级、岗位是组织架构管理的三个核心模块。部门采用树形结构,职级和岗位为简单 CRUD 模块,均继承 `BaseService`。用户表通过 `dept_id`、`position_id`、`level_id` 三个外键关联这三个模块。 ::: ## 部门管理 ### 部门模型 ```python # src/modules/dept/models.py # ============================================================ # 部门模型 # ============================================================ class Dept(BaseModel): """部门模型类""" # ============================================================ # 表名配置 # ============================================================ __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) ``` ### 删除前校验 部门删除时需检查是否存在下级部门或用户引用,避免组织架构悬空: ```python # src/modules/dept/service.py def _before_delete(self, ids) -> Optional[str]: id_list = parse_id_list(str(ids)) # 检查是否存在下级部门 if id_list and self.repo.filter( Dept.parent_id.in_(id_list), Dept.is_delete == 0 ).first(): return "存在下级部门,请先删除下级部门" # 检查是否存在用户引用 if id_list and user_repo.filter( User.dept_id.in_(id_list), User.is_delete == 0 ).first(): return "存在用户引用该部门,请先调整用户部门" return None ``` ## 职级管理 ### 职级模型 ```python # src/modules/level/models.py # ============================================================ # 职级模型 # ============================================================ class Level(BaseModel): """职级模型类""" # ============================================================ # 表名配置 # ============================================================ __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) ``` ### 表单验证 ```python # src/modules/level/schemas.py # ============================================================ # 职级表单类 # ============================================================ class LevelForm(BaseSchema): """职级创建/编辑表单""" name: str = Field(..., min_length=1, max_length=150, description="职级名称") status: int = Field(..., ge=1, le=2, description="职级状态:1-在用 2-停用") sort: int = Field(..., ge=0, le=99999, description="职级排序") # ============================================================ # 设置状态表单类 # ============================================================ class LevelStatusForm(BaseModel): """职级状态更新表单""" id: int = Field(..., gt=0, description="职级ID") status: int = Field(..., ge=1, le=2, description="职级状态:1-在用 2-停用") ``` ### 业务服务 职级模块继承 `BaseService`,声明差异点即可获得完整 CRUD 能力。删除前校验用户引用,支持 Excel 导入导出: ```python # src/modules/level/service.py class LevelService(BaseService(Level]): repo = level_repo model = Level page_like_fields = ('name',) page_eq_fields = ('status',) page_order_by = (('sort', 'asc'),) unique_fields = {'name': '职级名称不能重复'} def _before_delete(self, ids) -> Optional[str]: """删除前校验:存在用户引用该职级时禁止删除""" id_list = parse_id_list(str(ids)) if id_list and user_repo.filter(User.level_id.in_(id_list), User.is_delete == 0).first(): return "存在用户引用该职级,请先调整用户职级" return None level_service = LevelService() ``` ## 岗位管理 ### 岗位模型 ```python # src/modules/position/models.py # ============================================================ # 岗位模型 # ============================================================ class Position(BaseModel): """岗位模型类""" # ============================================================ # 表名配置 # ============================================================ __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) ``` ### 表单验证 ```python # src/modules/position/schemas.py # ============================================================ # 岗位表单类 # ============================================================ class PositionForm(BaseSchema): """岗位创建/编辑表单""" name: str = Field(..., min_length=1, max_length=150, description="岗位名称") status: int = Field(..., ge=1, le=2, description="岗位状态:1-在用 2-停用") sort: int = Field(..., ge=0, le=99999, description="岗位排序") # ============================================================ # 设置状态表单类 # ============================================================ class PositionStatusForm(BaseModel): """岗位状态更新表单""" id: int = Field(..., gt=0, description="岗位ID") status: int = Field(..., ge=1, le=2, description="岗位状态:1-在用 2-停用") ``` ### 数据访问层 ```python # src/modules/position/repository.py # +====================================================================== # | 模块: 岗位数据访问层 # | 说明: 封装岗位模型的数据访问操作 # +====================================================================== from position.models import Position from core.base_repository import BaseRepository # ============================================================ # 岗位仓库类 # ============================================================ class PositionRepository(BaseRepository(Position]): """岗位数据仓库,继承基础仓库获得通用 CRUD 能力""" pass # ============================================================ # 仓库实例 # ============================================================ position_repo = PositionRepository() ``` ### 业务服务 ```python # src/modules/position/service.py # ============================================================ # 岗位业务服务类 # ============================================================ class PositionService(BaseService(Position]): """岗位业务服务类,继承基础服务获得通用 CRUD 能力""" # 数据访问与模型 repo = position_repo model = Position # 分页差异点 page_like_fields = ('name',) page_eq_fields = ('status',) page_order_by = (('sort', 'asc'),) # 唯一性校验 unique_fields = {'name': '岗位名称不能重复'} # ============================================================ # 删除前校验 # ============================================================ def _before_delete(self, ids) -> Optional[str]: """删除前校验:存在用户引用该岗位时禁止删除,避免用户岗位悬空""" id_list = parse_id_list(str(ids)) if id_list and user_repo.filter(User.position_id.in_(id_list), User.is_delete == 0).first(): return "存在用户引用该岗位,请先调整用户岗位" return None # ============================================================ # 获取岗位数据列表 # ============================================================ def get_position_list(self, request): """获取岗位数据列表""" # 与 /page 一致的筛选条件:name 模糊、status 精确;保留软删过滤与排序 query = self._apply_page_filters(self.repo.filter_by(), request) query = query.order_by(self.model.sort.asc()) # 返回结果 return [v.to_dict() for v in query.all()] # ============================================================ # 模块级单例 # ============================================================ position_service = PositionService() ``` ## 通用 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` | 设置状态 | :::tip Excel 导入导出 职级模块支持 Excel 导入导出(`import_level` / `export_level`),导入时逐行校验并返回成功/失败条数。岗位模块结构相同,可按需添加导入导出功能。 ::: ## 总结 组织架构模块具备以下特点: ``` 1. 部门树形:parent_id 形成多级部门结构,删除前校验下级部门和用户引用 2. 职级/岗位:简单 CRUD,继承 BaseService,声明差异点即可 3. 表单验证:Pydantic Field 约束(ge/le/min_length/max_length) 4. 数据仓库:BaseRepository 提供通用 CRUD 能力,模块级单例 5. 用户关联:用户表通过 dept_id、position_id、level_id 关联 6. 删除保护:_before_delete 钩子防止引用悬空 ```