From 895983ac2095dac6dcdb0ca9b298e1bc7d3e8655 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Mon, 27 Jul 2026 09:50:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(data-process):=20=E8=BF=94=E5=9B=9E?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E6=96=87=E6=A1=A3=E6=95=B0=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/modules/data_process/store.py | 23 +++++---- backend/tests/test_data_process_api.py | 35 ++++++++++++- backend/tests/test_data_process_store.py | 61 +++++++++++++++++++++++ 3 files changed, 109 insertions(+), 10 deletions(-) diff --git a/backend/app/modules/data_process/store.py b/backend/app/modules/data_process/store.py index 79bab2e..cd547b1 100644 --- a/backend/app/modules/data_process/store.py +++ b/backend/app/modules/data_process/store.py @@ -220,34 +220,39 @@ class DataProcessStore: tenant_id: str | None = None, project_id: str | None = None, ) -> dict[str, Any]: - clauses = ["deleted_at IS NULL"] + clauses = ["task.deleted_at IS NULL"] params: list[Any] = [] if keyword: - clauses.append("(name ILIKE %s OR COALESCE(description, '') ILIKE %s)") + clauses.append("(task.name ILIKE %s OR COALESCE(task.description, '') ILIKE %s)") pattern = f"%{keyword.strip()}%" params.extend([pattern, pattern]) if status: - clauses.append("status = %s") + clauses.append("task.status = %s") params.append(status) if process_type: - clauses.append("process_type = %s") + clauses.append("task.process_type = %s") params.append(process_type) if tenant_id: - clauses.append("tenant_id = %s") + clauses.append("task.tenant_id = %s") params.append(tenant_id) if project_id: - clauses.append("project_id = %s") + clauses.append("task.project_id = %s") params.append(project_id) where = " AND ".join(clauses) with self.connect() as conn: total = conn.execute( - f"SELECT COUNT(*) AS count FROM data_process_tasks WHERE {where}", params + f"SELECT COUNT(*) AS count FROM data_process_tasks task WHERE {where}", + params, ).fetchone()["count"] rows = conn.execute( f""" - SELECT * FROM data_process_tasks + SELECT task.*, + (SELECT COUNT(*) FROM data_process_source_files source_file + WHERE source_file.task_id=task.id + AND source_file.deleted_at IS NULL) AS source_file_count + FROM data_process_tasks task WHERE {where} - ORDER BY created_at DESC, id DESC + ORDER BY task.created_at DESC, task.id DESC LIMIT %s OFFSET %s """, [*params, page_size, (page - 1) * page_size], diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index ea506dd..4e98305 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -37,7 +37,13 @@ class FakeDataProcessStore: return f"{prefix}_{self.sequence}" def list_tasks(self, *, page: int, page_size: int, **filters: Any) -> dict[str, Any]: - items = list(self.tasks.values()) + items = [ + { + **item, + "source_file_count": len(self.sources.get(str(item["id"]), [])), + } + for item in self.tasks.values() + ] for field in ("status", "process_type", "tenant_id", "project_id"): if filters.get(field): items = [item for item in items if item.get(field) == filters[field]] @@ -673,6 +679,33 @@ def test_data_process_full_contract_without_database(tmp_path: Path) -> None: ) +def test_task_list_exposes_document_and_generation_counts( + tmp_path: Path, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = client.post( + "/modelTF/data-process", + json={"name": "列表契约", "process_type": "unstructured", "config": {}}, + ).json()["data"]["id"] + store.sources[task_id] = [{"id": "source-1"}, {"id": "source-2"}] + store.tasks[task_id].update( + { + "status": "pending", + "output_count": 17, + "output_dataset_id": None, + } + ) + + response = client.get("/modelTF/data-process") + + assert response.status_code == 200 + item = response.json()["data"]["items"][0] + assert item["status"] == "pending" + assert item["source_file_count"] == 2 + assert item["output_count"] == 17 + assert item["output_dataset_id"] is None + + def test_preview_build_replaces_only_selected_files_and_reports_file_counts( tmp_path: Path, ) -> None: diff --git a/backend/tests/test_data_process_store.py b/backend/tests/test_data_process_store.py index a673f73..991c0d8 100644 --- a/backend/tests/test_data_process_store.py +++ b/backend/tests/test_data_process_store.py @@ -316,6 +316,54 @@ class _TaskDetailStore(DataProcessStore): yield self._conn +class _TaskListConnection: + def __init__(self) -> None: + self.task = { + **_regeneration_task( + status="pending", + output_dataset_id=None, + output_count=17, + ), + "created_at": "2026-07-25T18:00:00Z", + "deleted_at": None, + } + self.sources = [ + {"task_id": "task-1", "deleted_at": None}, + {"task_id": "task-1", "deleted_at": None}, + {"task_id": "task-1", "deleted_at": "2026-07-26T00:00:00Z"}, + ] + + def execute(self, sql: str, params: Any = None) -> _Result: + normalized = " ".join(sql.split()) + if normalized.startswith("SELECT COUNT(*) AS count FROM data_process_tasks task"): + assert "task.deleted_at IS NULL" in normalized + return _Result(row={"count": 1}) + assert normalized.startswith("SELECT task.*") + assert "source_file.task_id=task.id" in normalized + assert "source_file.deleted_at IS NULL" in normalized + source_file_count = sum( + item["task_id"] == self.task["id"] and item["deleted_at"] is None + for item in self.sources + ) + return _Result( + rows=[ + { + **self.task, + "source_file_count": source_file_count, + } + ] + ) + + +class _TaskListStore(DataProcessStore): + def __init__(self, conn: _TaskListConnection) -> None: + self._conn = conn + + @contextmanager + def connect(self) -> Iterator[_TaskListConnection]: + yield self._conn + + def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None: decoded = _decode_row( { @@ -461,6 +509,19 @@ def _regeneration_task(**updates: Any) -> dict[str, Any]: return task +def test_list_tasks_exposes_source_and_generation_counts() -> None: + page = _TaskListStore(_TaskListConnection()).list_tasks(page=1, page_size=20) + + assert page["total"] == 1 + assert page["page"] == 1 + assert page["page_size"] == 20 + item = page["items"][0] + assert item["status"] == "pending" + assert item["source_file_count"] == 2 + assert item["output_count"] == 17 + assert item["output_dataset_id"] is None + + def _legacy_published_datasets(task_id: str = "task-1") -> list[dict[str, Any]]: specs = ( ("dataset_train", "制度问答-训练集", "train", "train", 22),