请求解析工具

说明

src/common/utils/request.py 提供请求数据解析相关的工具函数，包括批量 ID 解析、SQL LIKE 安全处理、分页参数解析等。这些工具函数在各业务模块中广泛使用。

parse_batch_ids

从请求体解析批量删除的 ID 数组，校验并转为整数列表。

from flask import request
from src.common.utils.request import parse_batch_ids

def parse_batch_ids(request) -> list:
    """
    从请求体解析 ID 数组
    请求体格式: [1, 2, 3]
    返回: [1, 2, 3]

    示例:
    >>> parse_batch_ids(request)
    [1, 2, 3]
    """
    ids = request.get_json()
    if not ids or not isinstance(ids, list) or len(ids) == 0:
        raise BizError("参数错误：ids必须为非空数组")
    try:
        id_list = [int(id_val) for id_val in ids]
    except (ValueError, TypeError):
        raise BizError("参数错误：数组元素必须为数字")
    return id_list

使用示例：

@position_blue.route('/batchDelete', methods=['DELETE'])
@check_permission('sys:position:batchDelete')
def batch_delete():
    from src.modules.system.position.service import delete_positions
    id_list = parse_batch_ids(request)
    return delete_positions(id_list)

parse_request_body

解析请求体 JSON 数据，并使用 WTForms 进行校验。

from flask import request
from src.common.utils.request import parse_request_body

def parse_request_body(request, form_class):
    """
    解析请求体并校验

    参数:
        request: Flask 请求对象
        form_class: WTForms 表单类

    返回:
        dict: 校验通过的数据

    异常:
        BizError: 校验失败时抛出
    """
    data = request.get_json()
    if not data:
        raise BizError("请求体不能为空")

    form = form_class(data=data)
    if not form.validate():
        # 提取第一个错误信息
        errors = form.errors
        first_error = list(errors.values())[0][0] if errors else "参数校验失败"
        raise BizError(first_error)

    return {field.name: field.data for field in form}

使用示例：

@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
def add():
    from src.modules.system.position.service import add_position
    from src.modules.system.position.schema import PositionForm
    data = parse_request_body(request, PositionForm)
    return add_position(data)

parse_pagination

安全解析分页参数 pageNo / pageSize。

from flask import request
from src.common.utils.request import parse_pagination

def parse_pagination(default_page_size: int = 10, max_page_size: int = 500) -> tuple:
    """
    安全解析分页参数

    参数:
        default_page_size: 默认每页数量
        max_page_size: 最大每页数量

    返回:
        tuple: (page, page_size)，page>=1，1<=page_size<=max_page_size

    示例:
    >>> parse_pagination()
    (1, 10)
    """
    page = request.args.get('pageNo', 1, type=int)
    page_size = request.args.get('pageSize', default_page_size, type=int)

    # 非数字、小于1、超上限等一律回退/夹取
    page = max(1, page)
    page_size = max(1, min(page_size, max_page_size))

    return page, page_size

使用示例：

def get_position_page():
    """获取岗位分页列表"""
    page, page_size = parse_pagination()
    # 使用 page 和 page_size 进行分页查询
    # ...

like_escape

转义 SQL LIKE 通配符，防止用户输入被当作通配符。

from src.common.utils.request import like_escape

def like_escape(value, escape_char='\\'):
    """
    转义 SQL LIKE 通配符

    示例:
    >>> like_escape('100%')
    '100\\%'
    >>> like_escape('test_data')
    'test\\_data'
    """
    escaped = str(value).replace(escape_char, escape_char * 2)
    escaped = escaped.replace('%', escape_char + '%')
    escaped = escaped.replace('_', escape_char + '_')
    return escaped

like_contains

构建安全的 LIKE 包含查询值（前后加 %）。

from src.common.utils.request import like_contains

def like_contains(value, escape_char='\\'):
    """
    构建安全的 LIKE 包含查询值

    SQLAlchemy 用法:
    query = query.filter(User.name.like(like_contains(name), escape='\\\\'))
    """
    return f'%{like_escape(value, escape_char)}%'

使用示例：

from src.common.utils.request import like_contains

def search_users(keyword):
    """搜索用户"""
    return UserRepository.find_list(
        like_filters={'username': keyword} if keyword else None,
    )

get_client_ip

获取客户端真实 IP 地址。

from flask import request
from src.common.utils.request import get_client_ip

def get_client_ip() -> str:
    """
    获取客户端真实 IP 地址

    支持反向代理（X-Forwarded-For）和直接连接
    """
    if request.headers.get('X-Forwarded-For'):
        return request.headers.get('X-Forwarded-For').split(',')[0].strip()
    return request.remote_addr or '127.0.0.1'

总结

请求解析工具函数覆盖了批量操作、模糊查询安全、分页参数等常见场景。parse_batch_ids 和 parse_pagination 通过防御性校验避免畸形参数导致 500 错误，like_escape 防止 LIKE 通配符注入，parse_request_body 使用 WTForms 进行参数校验。