From 0cfaf8c89467f9309314199ac5ec241c71c4af6b Mon Sep 17 00:00:00 2001 From: DjangoAdmin Dev Date: Fri, 7 Aug 2026 14:42:03 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20P4b=20position=20=E6=A0=87=E6=9D=86?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E6=9C=8D=E5=8A=A1=E5=B1=82=E6=94=B9=E9=80=A0?= =?UTF-8?q?=20+=20=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【服务层新范式】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 --- src/modules/system/position/repository.py | 15 ++ src/modules/system/position/router.py | 35 ++-- src/modules/system/position/service.py | 189 +++++++----------- tests/conftest.py | 13 +- tests/unit/test_modules/__init__.py | 1 + .../test_modules/test_position_service.py | 81 ++++++++ 6 files changed, 204 insertions(+), 130 deletions(-) create mode 100644 src/modules/system/position/repository.py create mode 100644 tests/unit/test_modules/__init__.py create mode 100644 tests/unit/test_modules/test_position_service.py diff --git a/src/modules/system/position/repository.py b/src/modules/system/position/repository.py new file mode 100644 index 0000000..929ce48 --- /dev/null +++ b/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 diff --git a/src/modules/system/position/router.py b/src/modules/system/position/router.py index 0f9d0bb..a6645e6 100644 --- a/src/modules/system/position/router.py +++ b/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 result + return R.ok(data=position.get_position_page(request.args)) # ============================================================ @@ -124,8 +123,11 @@ def add(): Returns: JSON: 操作结果 """ - result = position.add_position() - return result + json_data, err = parse_request_body() + 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() - return result + json_data, err = parse_request_body() + 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) - return result + count = position.delete_positions(id) + 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() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + position.update_position_status(json_data) + return R.ok(msg="设置成功") # ============================================================ diff --git a/src/modules/system/position/service.py b/src/modules/system/position/service.py index ab5e0d0..0005e93 100644 --- a/src/modules/system/position/service.py +++ b/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.core import response as R +from src.modules.system.position.repository import PositionRepository +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(request.args.get('pageNo', 1)) - page_size = int(request.args.get('pageSize', PAGE_SIZE)) + page_no = int(params.get('pageNo', 1)) + page_size = int(params.get('pageSize', PAGE_SIZE)) - # 应用筛选条件 - query = _apply_filters() + # 应用筛选条件 + query = _apply_filters(params) - # 记录总数 - total = query.count() - # 计算总页数 - 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="查询失败，请稍后重试") + # 分页查询（repository 统一封装分页计算） + position_list, total, pages = PositionRepository.paginate(query, page_no, page_size) + return _build_page_data(position_list, total, page_size, page_no, pages) -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 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 + # 参数有效性验证与查询统一由仓储处理 + return PositionRepository.get_by_id(position_id) # ============================================================ # 岗位添加 # ============================================================ -def add_position(): +def add_position(data: dict) -> None: """ 添加岗位 流程： - 1. 解析请求体JSON数据 - 2. 表单验证（名称唯一性、字段格式等） - 3. 创建岗位对象并保存到数据库 + 1. 表单验证（名称唯一性、字段格式等） + 2. 创建岗位对象并保存到数据库 - Returns: - R: 操作结果 + Args: + data: 请求体数据 dict + + Raises: + BizError: 表单验证失败时 """ - try: - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) + form = PositionForm(data=data) + if not form.validate(): + raise BizError(regular.get_err(form)) - # 表单验证 - form = PositionForm(data=json_data) - if not form.validate(): - err_msg = regular.get_err(form) - 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="添加失败，请稍后重试") + # 创建岗位 + position = Position(**form.data) + position.create_user = get_username() + position.save() # ============================================================ # 岗位更新 # ============================================================ -def update_position(): +def update_position(data: dict) -> None: """ 更新岗位 流程： - 1. 解析请求体JSON数据 - 2. 表单验证 - 3. 验证岗位是否存在（复用 get_info） - 4. 更新岗位信息 + 1. 表单验证 + 2. 验证岗位是否存在（复用 get_info） + 3. 更新岗位信息 - Returns: - R: 操作结果 + Args: + data: 请求体数据 dict + + Raises: + BizError: 表单验证失败 / 岗位不存在 / 更新失败 """ - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) - - # 表单验证 - form = PositionForm(data=json_data) + form = PositionForm(data=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 position: - return R.failed("岗位不存在") + if not get_info(position_id): + raise BizError("岗位不存在") # 更新岗位信息（移除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("更新失败") - - return R.ok(msg="更新成功") + raise BizError("更新失败") # ============================================================ # 岗位删除 # ============================================================ -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: - R: 操作结果 - """ - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) + Args: + data: 请求体数据 dict - # 表单验证 - form = PositionStatusForm(data=json_data) + Raises: + BizError: 表单验证失败 / 岗位不存在 / 设置失败 + """ + form = PositionStatusForm(data=data) if not form.validate(): - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + raise BizError(regular.get_err(form)) position_id = int(form.data['id']) status = int(form.data['status']) # 复用 get_info 方法验证岗位是否存在 - position = get_info(position_id) - if not position: - return R.failed("岗位不存在") + if not get_info(position_id): + raise BizError("岗位不存在") # 更新状态 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("设置失败") - - return R.ok(msg="设置成功") + raise BizError("设置失败") # ============================================================ diff --git a/tests/conftest.py b/tests/conftest.py index 412e23a..3252b7a 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/unit/test_modules/__init__.py b/tests/unit/test_modules/__init__.py new file mode 100644 index 0000000..0b2d69c --- /dev/null +++ b/tests/unit/test_modules/__init__.py @@ -0,0 +1 @@ +"""业务模块服务单元测试。""" diff --git a/tests/unit/test_modules/test_position_service.py b/tests/unit/test_modules/test_position_service.py new file mode 100644 index 0000000..be67e27 --- /dev/null +++ b/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]