refactor: P4b position 标杆模块服务层改造 + 单元测试

【服务层新范式】system/position 作为标杆：
- service 函数改「入参数据/出参数据」：get_position_page(params)->dict、
  add_position(data)/update_position(data)/update_position_status(data)->None、
  delete_positions(ids)->int
- 业务失败 raise BizError（不再 return R.failed），由全局处理器转 R.failed
- get_info 委托 PositionRepository（新增 repository.py，继承 BaseRepository）
- router 负责取参（query args / parse_request_body）与 R 包装

【单元测试】tests/unit/test_modules/test_position_service.py：
- 纯函数（_build_page_data）、BizError 契约（空数据/空ID/无效ID）、
  mock 批量删除（ID 解析验证）——无需数据库即可测试
- conftest flask_app fixture 增加 SECRET_KEY + 关闭 CSRF

【验收】12 passed（5 响应 + 7 position）；应用启动；flake8 零告警；
position 路由 401 拦截正常。服务层测试化范式验证可行

Co-Authored-By: Claude <noreply@anthropic.com>

src/modules/system/position/repository.py

@@ -0,0 +1,15 @@
+"""
+岗位仓储模块
+=============================================================
+
+岗位数据访问层，继承 BaseRepository 获得通用 CRUD 能力。
+"""
+
+from src.common.base_repository import BaseRepository
+from src.modules.system.position.models import Position
+
+
+class PositionRepository(BaseRepository):
+    """岗位仓储"""
+
+    model = Position

src/modules/system/position/router.py

@@ -55,12 +55,12 @@
 依赖模块： application.services.position - 岗位业务逻辑层
          application.middleware.check_permission - 权限校验中间件
 """
-from flask import Blueprint
+from flask import Blueprint, request
 
 from application.middleware.check_demo import check_demo
 from application.middleware.check_permission import check_permission
 from src.modules.system.position import service as position
-from src.common.utils.request import parse_batch_ids
+from src.common.utils.request import parse_batch_ids, parse_request_body
 
 from src.core import response as R
 
@@ -83,8 +83,7 @@ def page():
     Returns:
         JSON: 岗位分页数据，包含 records（数据列表）、total（总记录数）等
     """
-    result = position.get_position_page()
+    return R.ok(data=position.get_position_page(request.args))
-    return result
 
 
 # ============================================================
@@ -124,8 +123,11 @@ def add():
     Returns:
         JSON: 操作结果
     """
-    result = position.add_position()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    position.add_position(json_data)
+    return R.ok(msg="添加成功")
 
 
 # ============================================================
@@ -145,8 +147,11 @@ def update():
     Returns:
         JSON: 操作结果
     """
-    result = position.update_position()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    position.update_position(json_data)
+    return R.ok(msg="更新成功")
 
 
 # ============================================================
@@ -166,8 +171,8 @@ def delete(id):
     Returns:
         JSON: 操作结果
     """
-    result = position.delete_positions(id)
+    count = position.delete_positions(id)
-    return result
+    return R.ok(msg=f"本次共删除{count}条数据")
 
 
 # ============================================================
@@ -193,7 +198,8 @@ def batch_delete():
     id_str, err = parse_batch_ids()
     if err:
         return err
-    return position.delete_positions(id_str)
+    count = position.delete_positions(id_str)
+    return R.ok(msg=f"本次共删除{count}条数据")
 
 
 # ============================================================
@@ -213,8 +219,11 @@ def status():
     Returns:
         JSON: 操作结果
     """
-    result = position.update_position_status()
+    json_data, err = parse_request_body()
-    return result
+    if err:
+        return R.failed(msg=err)
+    position.update_position_status(json_data)
+    return R.ok(msg="设置成功")
 
 
 # ============================================================

src/modules/system/position/service.py

@@ -65,18 +65,17 @@
 import logging
 
 from datetime import datetime, timezone
-from flask import request
 from sqlalchemy import and_
 
 from src.common.constants.common import PAGE_SIZE
 from src.modules.system.position.schema import PositionForm, PositionStatusForm
 from src.modules.system.position.models import Position
-from src.core.database import db
+from src.modules.system.position.repository import PositionRepository
-from src.core import response as R
+from src.core.exceptions import BizError
 from src.common.utils import regular
 from src.common.utils.case_converter import convert_camel_to_snake
 from src.common.utils.model_utils import models_to_list, model_to_dict
-from src.common.utils.request import parse_request_body, parse_id_list, like_contains
+from src.common.utils.request import parse_id_list, like_contains
 from src.core.security import get_username
 
 # 获取日志记录器
@@ -87,19 +86,19 @@ logger = logging.getLogger(__name__)
 # 岗位分页查询
 # ============================================================
 
-def get_position_page():
+def get_position_page(params: dict) -> dict:
     """
     查询岗位分页数据
 
     Args:
-        request: Flask request对象，包含查询参数
+        params: 查询参数
             - pageNo: 当前页码（可选，默认为1）
             - pageSize: 每页记录数（可选，默认为PAGE_SIZE）
             - name: 岗位名称（可选，模糊查询）
             - status: 岗位状态（可选，精确匹配）
 
     Returns:
-        R: 包含分页数据的JSON响应，格式为：
+        dict: 分页数据，格式为：
             {
                 "records": 岗位列表,
                 "total": 总记录数,
@@ -108,40 +107,30 @@ def get_position_page():
                 "pages": 总页数
             }
     """
-    try:
+    page_no = int(params.get('pageNo', 1))
-        # 获取分页参数
+    page_size = int(params.get('pageSize', PAGE_SIZE))
-        page_no = int(request.args.get('pageNo', 1))
-        page_size = int(request.args.get('pageSize', PAGE_SIZE))
 
-        # 应用筛选条件
+    # 应用筛选条件
-        query = _apply_filters()
+    query = _apply_filters(params)
 
-        # 记录总数
+    # 分页查询（repository 统一封装分页计算）
-        total = query.count()
+    position_list, total, pages = PositionRepository.paginate(query, page_no, page_size)
-        # 计算总页数
+    return _build_page_data(position_list, total, page_size, page_no, pages)
-        pages = (total + page_size - 1) // page_size if total > 0 else 0
-
-        # 分页查询
-        position_list = query.limit(page_size).offset((page_no - 1) * page_size).all()
-        # 构建返回数据
-        page_data = _build_page_data(position_list, total, page_size, page_no, pages)
-        return R.ok(data=page_data)
-
-    except Exception as e:
-        logger.error(f"查询岗位分页数据异常: {str(e)}")
-        return R.failed(msg="查询失败，请稍后重试")
 
 
-def _apply_filters():
+def _apply_filters(params: dict):
     """
     应用查询筛选条件
 
-    根据请求参数动态构建查询条件：
+    根据查询参数动态构建查询条件：
         - 自动过滤已删除记录（is_delete == 0）
         - 支持岗位名称模糊查询
         - 支持岗位状态精确查询
         - 默认按排序值升序排列
 
+    Args:
+        params: 查询参数 dict
+
     Returns:
         Query: 应用筛选后的SQLAlchemy查询对象
     """
@@ -149,12 +138,12 @@ def _apply_filters():
     query = Position.query.filter(Position.is_delete == 0)
 
     # 岗位名称模糊筛选
-    name = request.args.get('name')
+    name = params.get('name')
     if name:
         query = query.filter(Position.name.like(like_contains(name), escape='\\'))
 
     # 岗位状态精确筛选
-    status = request.args.get('status')
+    status = params.get('status')
     if status:
         query = query.filter(Position.status == status)
 
@@ -247,98 +236,70 @@ def get_info(position_id: int):
         if position:
             print(position.name)
     """
-    # 参数有效性验证
+    # 参数有效性验证与查询统一由仓储处理
-    if not position_id or position_id <= 0:
+    return PositionRepository.get_by_id(position_id)
-        return None
-
-    try:
-        # 查询未删除的岗位
-        position = Position.query.filter(
-            and_(Position.id == position_id, Position.is_delete == 0)
-        ).first()
-        return position
-    except Exception as e:
-        logger.error(f"查询岗位信息异常，ID: {position_id}, 错误: {str(e)}")
-        return None
 
 
 # ============================================================
 # 岗位添加
 # ============================================================
 
-def add_position():
+def add_position(data: dict) -> None:
     """
     添加岗位
 
     流程：
-        1. 解析请求体JSON数据
+        1. 表单验证（名称唯一性、字段格式等）
-        2. 表单验证（名称唯一性、字段格式等）
+        2. 创建岗位对象并保存到数据库
-        3. 创建岗位对象并保存到数据库
 
-    Returns:
+    Args:
-        R: 操作结果
+        data: 请求体数据 dict
+
+    Raises:
+        BizError: 表单验证失败时
     """
-    try:
+    form = PositionForm(data=data)
-        # 获取参数
+    if not form.validate():
-        json_data, error = parse_request_body()
+        raise BizError(regular.get_err(form))
-        if error:
-            return R.failed(msg=error)
 
-        # 表单验证
+    # 创建岗位
-        form = PositionForm(data=json_data)
+    position = Position(**form.data)
-        if not form.validate():
+    position.create_user = get_username()
-            err_msg = regular.get_err(form)
+    position.save()
-            return R.failed(msg=err_msg)
-
-        # 创建岗位
-        position = Position(**form.data)
-        position.create_user = get_username()
-        position.save()
-
-        return R.ok(msg="添加成功")
-
-    except Exception as e:
-        logger.error(f"添加岗位异常: {str(e)}")
-        return R.failed(msg="添加失败，请稍后重试")
 
 
 # ============================================================
 # 岗位更新
 # ============================================================
 
-def update_position():
+def update_position(data: dict) -> None:
     """
     更新岗位
 
     流程：
-        1. 解析请求体JSON数据
+        1. 表单验证
-        2. 表单验证
+        2. 验证岗位是否存在（复用 get_info）
-        3. 验证岗位是否存在（复用 get_info）
+        3. 更新岗位信息
-        4. 更新岗位信息
 
-    Returns:
+    Args:
-        R: 操作结果
+        data: 请求体数据 dict
+
+    Raises:
+        BizError: 表单验证失败 / 岗位不存在 / 更新失败
     """
-    # 获取参数
+    form = PositionForm(data=data)
-    json_data, error = parse_request_body()
-    if error:
-        return R.failed(msg=error)
-
-    # 表单验证
-    form = PositionForm(data=json_data)
     if not form.validate():
-        return R.failed(msg=regular.get_err(form))
+        raise BizError(regular.get_err(form))
 
     # 获取岗位ID
     position_id = form.data.get('id')
     if not position_id or int(position_id) <= 0:
-        return R.failed("岗位ID无效")
+        raise BizError("岗位ID无效")
     position_id = int(position_id)
 
     # 复用 get_info 方法验证岗位是否存在
-    position = get_info(position_id)
+    if not get_info(position_id):
-    if not position:
+        raise BizError("岗位不存在")
-        return R.failed("岗位不存在")
 
     # 更新岗位信息（移除ID，避免更新主键）
     form_data = convert_camel_to_snake(form.data.copy())
@@ -350,71 +311,69 @@ def update_position():
     result = Position.query.filter_by(id=position_id).update(form_data)
 
     if not result:
-        return R.failed("更新失败")
+        raise BizError("更新失败")
-
-    return R.ok(msg="更新成功")
 
 
 # ============================================================
 # 岗位删除
 # ============================================================
 
-def delete_positions(position_ids):
+def delete_positions(position_ids: str) -> int:
     """
     删除岗位（批量删除）
 
-    删除
     支持逗号分隔的多个ID批量删除。
 
     Args:
         position_ids: 岗位ID字符串，多个ID用逗号分隔，如 "1,2,3"
 
     Returns:
-        R: 操作结果
+        int: 本次删除的记录数
+
+    Raises:
+        BizError: ID 无效时
     """
-    return Position.batch_delete_r(position_ids)
+    if not position_ids:
+        raise BizError("记录ID不存在")
+    id_list = parse_id_list(position_ids)
+    if not id_list:
+        raise BizError("无效的记录ID")
+    return Position.batch_delete(id_list)
 
 
 # ============================================================
 # 岗位状态设置
 # ============================================================
 
-def update_position_status():
+def update_position_status(data: dict) -> None:
     """
     设置岗位状态
 
     单独更新岗位的启用/禁用状态，不涉及其他字段。
 
-    Returns:
+    Args:
-        R: 操作结果
+        data: 请求体数据 dict
-    """
-    # 获取参数
-    json_data, error = parse_request_body()
-    if error:
-        return R.failed(msg=error)
 
-    # 表单验证
+    Raises:
-    form = PositionStatusForm(data=json_data)
+        BizError: 表单验证失败 / 岗位不存在 / 设置失败
+    """
+    form = PositionStatusForm(data=data)
     if not form.validate():
-        err_msg = regular.get_err(form)
+        raise BizError(regular.get_err(form))
-        return R.failed(msg=err_msg)
 
     position_id = int(form.data['id'])
     status = int(form.data['status'])
 
     # 复用 get_info 方法验证岗位是否存在
-    position = get_info(position_id)
+    if not get_info(position_id):
-    if not position:
+        raise BizError("岗位不存在")
-        return R.failed("岗位不存在")
 
     # 更新状态
     update_data = {"status": status, "update_time": datetime.now(tz=timezone.utc), "update_user": get_username()}
     result = Position.query.filter_by(id=position_id).update(update_data)
 
     if not result:
-        return R.failed("设置失败")
+        raise BizError("设置失败")
-
-    return R.ok(msg="设置成功")
 
 
 # ============================================================

tests/conftest.py

@@ -27,5 +27,14 @@ if ROOT not in sys.path:
 
 @pytest.fixture
 def flask_app():
-    """最小 Flask 应用实例（无数据库/中间件，仅提供应用上下文）。"""
+    """
-    return Flask(__name__)
+    最小 Flask 应用实例（无数据库/中间件，仅提供应用上下文）。
+
+    配置项说明：
+        - SECRET_KEY: FlaskForm CSRF 校验需要
+        - WTF_CSRF_ENABLED=False: 单元测试关闭 CSRF
+    """
+    app = Flask(__name__)
+    app.config['SECRET_KEY'] = 'test-secret-key'
+    app.config['WTF_CSRF_ENABLED'] = False
+    return app

tests/unit/test_modules/__init__.py

@@ -0,0 +1 @@
+"""业务模块服务单元测试。"""

tests/unit/test_modules/test_position_service.py

@@ -0,0 +1,81 @@
+"""
+岗位服务单元测试（P4 服务层改造标杆）
+=============================================================
+
+验证 P4 改造后的服务层契约：
+    - 服务函数「入参数据、出参数据」
+    - 业务失败通过 raise BizError 表达（不再 return R.failed）
+    - 纯函数与可 mock 路径无需数据库即可测试
+"""
+
+import pytest
+
+from src.core.exceptions import BizError
+from src.modules.system.position import service as position_service
+
+
+class TestBuildPageData:
+    """分页数据构建（纯函数）"""
+
+    def test_shape(self):
+        data = position_service._build_page_data([], 0, 10, 1, 0)
+        assert data == {'records': [], 'total': 0, 'size': 10, 'current': 1, 'pages': 0}
+
+
+class TestAddPosition:
+    """添加岗位"""
+
+    def test_empty_data_raises_bizerror(self, flask_app):
+        """空数据触发表单验证失败 → BizError（FlaskForm 需应用上下文）"""
+        with flask_app.app_context():
+            with pytest.raises(BizError):
+                position_service.add_position({})
+
+
+class TestUpdatePosition:
+    """更新岗位"""
+
+    def test_empty_data_raises_bizerror(self, flask_app):
+        with flask_app.app_context():
+            with pytest.raises(BizError):
+                position_service.update_position({})
+
+
+class TestUpdateStatus:
+    """状态设置"""
+
+    def test_empty_data_raises_bizerror(self, flask_app):
+        with flask_app.app_context():
+            with pytest.raises(BizError):
+                position_service.update_position_status({})
+
+
+class TestDeletePositions:
+    """批量删除"""
+
+    def test_empty_raises(self):
+        """空 ID 触发 BizError"""
+        with pytest.raises(BizError, match="记录ID不存在"):
+            position_service.delete_positions('')
+
+    def test_invalid_ids_raises(self):
+        """非数字 ID 解析为空 → BizError"""
+        with pytest.raises(BizError, match="无效的记录ID"):
+            position_service.delete_positions('abc')
+
+    def test_parse_and_delete(self, monkeypatch):
+        """ID 解析后调用模型批量删除，返回删除计数"""
+        from src.modules.system.position.models import Position
+
+        captured = {}
+
+        def fake_batch_delete(cls, ids):
+            captured['ids'] = ids
+            return len(ids)
+
+        monkeypatch.setattr(Position, 'batch_delete', classmethod(fake_batch_delete))
+
+        count = position_service.delete_positions('1,2,3')
+
+        assert count == 3
+        assert captured['ids'] == [1, 2, 3]