124 lines
3.4 KiB
Markdown
124 lines
3.4 KiB
Markdown
# 请求解析工具
|
||
|
||
:::tip 说明
|
||
`src/utils/request.py` 提供请求数据解析相关的工具函数，包括批量 ID 解析、SQL LIKE 安全处理、分页参数解析等。这些工具函数在各业务模块中广泛使用。
|
||
:::
|
||
|
||
## parse_batch_ids
|
||
|
||
从请求体解析批量删除的 ID 数组，校验并转为逗号分隔字符串。
|
||
|
||
```python
|
||
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
|
||
```
|
||
|
||
使用示例：
|
||
|
||
```python
|
||
@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)
|
||
```
|
||
|
||
:::tip 温馨提示
|
||
`parse_batch_ids` 是异步函数（需要 `await request.json()`），因此批量删除端点必须声明为 `async def`。
|
||
:::
|
||
|
||
## parse_id_list
|
||
|
||
将逗号分隔的 ID 字符串解析为整数列表。
|
||
|
||
```python
|
||
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 通配符，防止用户输入被当作通配符。
|
||
|
||
```python
|
||
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 包含查询值（前后加 `%`）。
|
||
|
||
```python
|
||
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`。
|
||
|
||
```python
|
||
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 数据。
|
||
|
||
```python
|
||
async def get_json_data(request, required=True):
|
||
"""获取请求中的 JSON 数据，失败返回 None 或空字典"""
|
||
```
|
||
|
||
## parse_request_body
|
||
|
||
解析请求体 JSON 数据，返回 `(data, error)` 元组。
|
||
|
||
```python
|
||
async def parse_request_body(request):
|
||
"""成功时 data 为字典，error 为 None；失败时 data 为 None，error 为错误描述"""
|
||
```
|
||
|
||
## 总结
|
||
|
||
请求解析工具函数覆盖了批量操作、模糊查询安全、分页参数等常见场景。`parse_batch_ids` 和 `parse_pagination` 通过防御性校验避免畸形参数导致 500 错误，`like_escape` 防止 LIKE 通配符注入。
|