feat(backend): add list export column filter and row-count guards

Shared helpers enforce empty/over-limit export rules and allow
clients to request visible columns by field key.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-21 16:52:50 +08:00
1 parent d26fb23fcf
commit 4a194aa25e
2 files changed
+75

No files matched your search

+32
View File
@@ -13,6 +13,12 @@ from app.models.models import SysDictData
CellFormatter = Callable[[Any], Any]
EXPORT_MAX_ROWS = 50000
class ExportValidationError(ValueError):
"""导出业务校验失败(空数据 / 超限等),路由层转为 400。"""
@dataclass(frozen=True)
class ExportColumn:
@@ -121,6 +127,29 @@ def workbook_to_streaming_response(
)
def filter_export_columns(
all_columns: Sequence[ExportColumn],
requested: Optional[Sequence[str]],
) -> List[ExportColumn]:
"""按前端可见列 prop 裁剪;未知 field 忽略;requested 为空则返回全部默认列。"""
if not requested:
return list(all_columns)
by_field = {c.field: c for c in all_columns}
out: List[ExportColumn] = []
for key in requested:
col = by_field.get(str(key))
if col is not None:
out.append(col)
return out if out else list(all_columns)
def assert_export_row_count(count: int) -> None:
if count <= 0:
raise ExportValidationError("没有可导出的数据")
if count > EXPORT_MAX_ROWS:
raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条,请缩小筛选范围")
def export_rows_to_xlsx_response(
db: Session,
*,
@@ -132,6 +161,9 @@ def export_rows_to_xlsx_response(
include_legacy_filename: bool = True,
) -> StreamingResponse:
"""CRM 列表导出统一入口:表头 + 行数据 + 字典 label 转换 + 文件流响应。"""
assert_export_row_count(len(rows))
if not columns:
raise ExportValidationError("没有可导出的列")
resolver = DictLabelResolver(db, columns, tenant_no=tenant_no)
wb = _build_workbook(sheet_title, columns, rows, resolver)
return workbook_to_streaming_response(
@@ -0,0 +1,43 @@
import pytest
from app.utils.excel_export import (
EXPORT_MAX_ROWS,
ExportColumn,
ExportValidationError,
filter_export_columns,
assert_export_row_count,
)
COLS = [
ExportColumn(title="账号", field="userNo"),
ExportColumn(title="姓名", field="nickName"),
ExportColumn(title="状态", field="status", dict_type="sys_status"),
]
def test_filter_export_columns_keeps_order_and_ignores_unknown():
out = filter_export_columns(COLS, ["status", "unknown", "userNo"])
assert [c.field for c in out] == ["status", "userNo"]
def test_filter_export_columns_none_returns_all():
out = filter_export_columns(COLS, None)
assert [c.field for c in out] == ["userNo", "nickName", "status"]
def test_assert_export_row_count_empty():
with pytest.raises(ExportValidationError) as ei:
assert_export_row_count(0)
assert "没有可导出的数据" in str(ei.value)
def test_assert_export_row_count_over_limit():
with pytest.raises(ExportValidationError) as ei:
assert_export_row_count(EXPORT_MAX_ROWS + 1)
assert "50000" in str(ei.value)
def test_assert_export_row_count_ok():
assert_export_row_count(1)
assert_export_row_count(EXPORT_MAX_ROWS)