请求解析工具

说明

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

parse_batch_ids

从请求体解析批量删除的 ID 数组，校验并转为逗号分隔字符串。

async def parse_batch_ids(request):
    """
    请求体格式: [1, 2, 3]
    返回: (id_str, error_response)
    成功时 id_str 为 "1,2,3"，error_response 为 None
    失败时 id_str 为 None，error_response 为 R.failed 响应
    """
    ids = await request.json()
    if not ids or not isinstance(ids, list) or len(ids) == 0:
        return None, R.failed("参数错误：ids必须为非空数组")
    try:
        id_list = [int(id_val) for id_val in ids]
    except (ValueError, TypeError):
        return None, R.failed("参数错误：数组元素必须为数字")
    return ','.join(str(id_val) for id_val in id_list), None

使用示例：

@router.delete('/batchDelete')
async def batch_delete(request: Request):
    id_str, err = await parse_batch_ids(request)
    if err:
        return err
    return position_service.batch_delete_by_ids(id_str)

温馨提示

parse_batch_ids 是异步函数（需要 await request.json()），因此批量删除端点必须声明为 async def。

parse_id_list

将逗号分隔的 ID 字符串解析为整数列表。

def parse_id_list(id_str: str, separator: str = ',') -> list:
    """
    示例:
    >>> parse_id_list("1,2,3")
    [1, 2, 3]
    >>> parse_id_list("1,2,abc,3")
    [1, 2, 3]  # 自动跳过非数字项
    """

like_escape

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

def like_escape(value, escape_char='\\'):
    """
    示例:
    >>> 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 包含查询值（前后加 %）。

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

parse_pagination

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

def parse_pagination(request, default_page_size: int = PAGE_SIZE, max_page_size: int = 500) -> tuple:
    """
    非数字、小于1、超上限等一律回退/夹取
    返回: (page, page_size)，page>=1，1<=page_size<=max_page_size
    """

get_json_data

从 FastAPI 请求中解析 JSON 数据。

async def get_json_data(request, required=True):
 """获取请求中的 JSON 数据，失败返回 None 或空字典"""

parse_request_body

解析请求体 JSON 数据，返回 (data, error) 元组。

async def parse_request_body(request):
 """成功时 data 为字典，error 为 None；失败时 data 为 None，error 为错误描述"""

总结

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