Become a sponsor
完整代码清单
本章汇总 Position 岗位模块涉及的全部文件，便于开发时快速查阅和复制。
后端文件清单
1. 数据模型 src/modules/system/position/models.py
python
"""
岗位模型模块
=============================================================
岗位模型定义，包含名称、排序、状态等属性。
"""
from sqlalchemy import text
from src.core.database import db
from src.common.base_db import BaseDB
from src.common.base_model import BaseModel
from config.database import DB_PREFIX
# ============================================================
# 岗位模型
# ============================================================
class Position(BaseModel, BaseDB):
"""岗位模型类"""
# ============================================================
# 表名配置
# ============================================================
__tablename__ = DB_PREFIX + "position"
__table_comment__ = "岗位表"
# ============================================================
# 字段定义
# ============================================================
# 岗位名称
name = db.Column(db.String(255), nullable=False, index=True, comment="岗位名称")
# 岗位状态：0-停用 1-启用
status = db.Column(db.Integer, default=0, server_default=text('0'), index=True, comment="岗位状态：0-停用 1-启用")
# 岗位排序
sort = db.Column(db.Integer, default=0, server_default=text('0'), comment="岗位排序")
# ============================================================
# 内置方法
# ============================================================
def __repr__(self):
"""返回岗位名称作为字符串表示"""
return '岗位：{}'.format(self.name)2. 表单验证 src/modules/system/position/schema.py
python
"""
岗位表单验证模块
=============================================================
提供岗位相关的表单验证功能，包括：
- 岗位添加/更新表单验证
- 岗位状态设置表单验证
"""
from wtforms import ValidationError, 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):
"""
岗位表单验证类
Attributes:
id: 岗位ID（更新时必填，添加时可选）
name: 岗位名称（必填，最长150字符，不可重复）
status: 岗位状态（必填，1-在用，2-停用）
sort: 岗位排序（可选，0-99999之间的整数）
"""
id = IntegerField(
label='岗位ID',
validators=[Optional(), NumberRange(min=1, message='岗位ID必须大于0')],
)
name = StringField(
label='岗位名称',
validators=[
DataRequired(message='岗位名称不能为空'),
Length(max=150, message='岗位名称长度不得超过150个字符')
],
)
status = IntegerField(
label='岗位状态',
validators=[
DataRequired(message='岗位状态不能为空'),
NumberRange(min=1, max=2, message='岗位状态值在1~2之间')
],
)
sort = IntegerField(
label='岗位排序',
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(
label='岗位ID',
validators=[DataRequired(message='岗位ID不能为空'), NumberRange(min=1, message='岗位ID必须大于0')],
)
status = IntegerField(
label='岗位状态',
validators=[
DataRequired(message='岗位状态不能为空'),
NumberRange(min=1, max=2, message='岗位状态值在1~2之间')
],
)3. 数据访问层 src/modules/system/position/repository.py
python
"""
岗位仓储模块
=============================================================
岗位数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
"""
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):
"""
岗位分页查询
Args:
params: 查询参数字典（pageNo/pageSize/name/status）
Returns:
tuple: (items, total, pages)
"""
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):
"""
查询启用的岗位列表
Args:
params: 查询参数字典，可包含 name 字段
Returns:
list: 启用状态的岗位实例列表
"""
params = params or {}
return cls.find_list(
filters={'status': 1},
like_filters={'name': params.get('name')},
order_by='sort',
)4. 业务逻辑层 src/modules/system/position/service.py
python
"""
岗位管理服务模块
=============================================================
该模块实现了岗位管理相关的所有业务逻辑。
"""
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} 已被用户引用，无法删除")
# ============================================================
# 模块级代理函数：保持 router 调用兼容
# ============================================================
get_position_page = PositionService.get_page
get_position_detail = PositionService.get_detail
get_info = PositionService.get_info
add_position = PositionService.add
update_position = PositionService.update
delete_positions = PositionService.remove
update_position_status = PositionService.update_status
get_position_list = PositionService.get_list5. 路由控制器 src/modules/system/position/router.py
python
"""
岗位管理控制器模块
=============================================================
提供岗位（Position）相关的API接口。
"""
from flask import Blueprint, request
from src.middleware import check_demo, operation_log
from src.core.permission import check_permission
from src.modules.system.position import service as position
from src.common.utils.request import parse_batch_ids, parse_request_body
from src.core import response as R
position_blue = Blueprint('position', __name__, url_prefix='/position')
@position_blue.route('/page', methods=["GET"])
@check_permission("sys:position:page")
@operation_log("岗位管理", "查询")
def page():
"""查询岗位分页数据"""
return R.ok(data=position.get_position_page(request.args))
@position_blue.route('/detail/<int:id>', methods=["GET"])
@check_permission("sys:position:detail")
@operation_log("岗位管理", "详情")
def detail(id):
"""查询岗位详情"""
data = position.get_position_detail(id)
return R.ok(data=data)
@position_blue.route('/add', methods=["POST"])
@check_permission("sys:position:add")
@check_demo
@operation_log("岗位管理", "新增")
def add():
"""添加岗位"""
json_data, err = parse_request_body()
if err:
return R.failed(msg=err)
position.add_position(json_data)
return R.ok(msg="添加成功")
@position_blue.route('/update', methods=["PUT"])
@check_permission("sys:position:update")
@check_demo
@operation_log("岗位管理", "修改")
def update():
"""更新岗位"""
json_data, err = parse_request_body()
if err:
return R.failed(msg=err)
position.update_position(json_data)
return R.ok(msg="更新成功")
@position_blue.route('/delete/<string:id>', methods=["DELETE"])
@check_permission("sys:position:delete")
@check_demo
@operation_log("岗位管理", "删除")
def delete(id):
"""删除岗位"""
count = position.delete_positions(id)
return R.ok(msg=f"本次共删除{count}条数据")
@position_blue.route('/batchDelete', methods=["DELETE"])
@check_permission("sys:position:batchDelete")
@check_demo
@operation_log("岗位管理", "批量删除")
def batch_delete():
"""批量删除岗位"""
id_str, err = parse_batch_ids()
if err:
return err
count = position.delete_positions(id_str)
return R.ok(msg=f"本次共删除{count}条数据")
@position_blue.route('/status', methods=["PUT"])
@check_permission("sys:position:status")
@check_demo
@operation_log("岗位管理", "状态变更")
def status():
"""设置岗位状态"""
json_data, err = parse_request_body()
if err:
return R.failed(msg=err)
position.update_position_status(json_data)
return R.ok(msg="设置成功")
@position_blue.route('/list', methods=["GET"])
@check_permission("sys:position:list")
@operation_log("岗位管理", "查询")
def get_position_list():
"""获取岗位数据列表（无分页）"""
result = position.get_position_list(request.args)
return R.ok(data=result)6. 路由注册 src/routes/v1.py
python
from src.modules.system.position.router import position_blue
# ... 其他模块导入
BLUEPRINTS = [
# 系统管理模块
user_blue, role_blue, menu_blue, dept_blue, position_blue, level_blue,
# ... 其他模块
]API 接口清单
| 方法 | 路径 | 权限节点 | 说明 |
|---|---|---|---|
| GET | /api/v1/position/page | sys:position:page | 分页查询 |
| GET | /api/v1/position/detail/<id> | sys:position:detail | 详情查询 |
| GET | /api/v1/position/list | sys:position:list | 列表查询（无分页） |
| POST | /api/v1/position/add | sys:position:add | 新增 |
| PUT | /api/v1/position/update | sys:position:update | 更新 |
| DELETE | /api/v1/position/delete/<id> | sys:position:delete | 单条删除 |
| DELETE | /api/v1/position/batchDelete | sys:position:batchDelete | 批量删除 |
| PUT | /api/v1/position/status | sys:position:status | 状态变更 |