Become a sponsor

部门/职级/岗位
部门、职级、岗位是组织架构管理的三个核心模块。部门采用树形结构,职级和岗位为简单 CRUD 模块,均继承 BaseService。用户表通过 dept_id、position_id、level_id 三个外键关联这三个模块。
# src/modules/system/dept/models.py
# ============================================================
# 部门模型
# ============================================================
class Dept(base_model, base_db):
"""部门模型类"""
# ============================================================
# 表名配置
# ============================================================
__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
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# src/modules/system/level/models.py
# ============================================================
# 职级模型
# ============================================================
class Level(base_model, base_db):
"""职级模型类"""
# ============================================================
# 表名配置
# ============================================================
__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/schemas.py
# ============================================================
# 职级表单类
# ============================================================
class LevelForm(BaseForm):
"""职级创建/编辑表单"""
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 导入导出:
# src/modules/system/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()# src/modules/system/position/models.py
# ============================================================
# 岗位模型
# ============================================================
class Position(base_model, base_db):
"""岗位模型类"""
# ============================================================
# 表名配置
# ============================================================
__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/schemas.py
# ============================================================
# 岗位表单类
# ============================================================
class PositionForm(BaseForm):
"""岗位创建/编辑表单"""
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-停用")# src/modules/system/position/repository.py
# +======================================================================
# | 模块: 岗位数据访问层
# | 说明: 封装岗位模型的数据访问操作
# +======================================================================
from modules.system.position.models import Position
from core.base_repository import BaseRepository
# ============================================================
# 岗位仓库类
# ============================================================
class PositionRepository(BaseRepository[Position]):
"""岗位数据仓库,继承基础仓库获得通用 CRUD 能力"""
pass
# ============================================================
# 仓库实例
# ============================================================
position_repo = PositionRepository(Position)# src/modules/system/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()职级和岗位模块继承 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. 表单验证:Pydantic Field 约束(ge/le/min_length/max_length)
4. 数据仓库:BaseRepository 提供通用 CRUD 能力,模块级单例
5. 用户关联:用户表通过 dept_id、position_id、level_id 关联
6. 删除保护:_before_delete 钩子防止引用悬空