211 lines
5.3 KiB
Markdown
211 lines
5.3 KiB
Markdown
# 请求解析工具
|
||
|
||
:::tip 说明
|
||
`src/common/utils/request.py` 提供请求数据解析相关的工具函数，包括批量 ID 解析、SQL LIKE 安全处理、分页参数解析等。这些工具函数在各业务模块中广泛使用。
|
||
:::
|
||
|
||
## parse_batch_ids
|
||
|
||
从请求体解析批量删除的 ID 数组，校验并转为整数列表。
|
||
|
||
```python
|
||
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
|
||
```
|
||
|
||
使用示例：
|
||
|
||
```python
|
||
@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 进行校验。
|
||
|
||
```python
|
||
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}
|
||
```
|
||
|
||
使用示例：
|
||
|
||
```python
|
||
@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`。
|
||
|
||
```python
|
||
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
|
||
```
|
||
|
||
使用示例：
|
||
|
||
```python
|
||
def get_position_page():
|
||
"""获取岗位分页列表"""
|
||
page, page_size = parse_pagination()
|
||
# 使用 page 和 page_size 进行分页查询
|
||
# ...
|
||
```
|
||
|
||
## like_escape
|
||
|
||
转义 SQL LIKE 通配符，防止用户输入被当作通配符。
|
||
|
||
```python
|
||
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 包含查询值（前后加 `%`）。
|
||
|
||
```python
|
||
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)}%'
|
||
```
|
||
|
||
使用示例：
|
||
|
||
```python
|
||
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 地址。
|
||
|
||
```python
|
||
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 进行参数校验。
|