From 6ee6ba9726f139c3e88e06a282fa8a21624e22e5 Mon Sep 17 00:00:00 2001 From: muhammadshayan124 Date: Sun, 5 Jul 2026 16:30:49 +0500 Subject: [PATCH] feat: add debug.list_tasks to expose GET /v1/tasks (#2059) Adds a list_tasks() method to the sync and async debug namespace that calls the server's distributed-tasks endpoint, so users can inspect long-running background operations (e.g. reindexing) tracked across cluster nodes without shelling out to curl. Includes DistributedTask/DistributedTaskUnit pydantic models mirroring the server's DistributedTasks OpenAPI schema, mock tests covering the populated and empty-response cases, and integration test stubs following the existing debug test conventions. --- integration/test_client_debug.py | 17 ++++++++++ mock_tests/test_debug.py | 56 ++++++++++++++++++++++++++++++++ weaviate/classes/debug.py | 4 ++- weaviate/debug/async_.pyi | 5 +-- weaviate/debug/executor.py | 25 ++++++++++++-- weaviate/debug/sync.pyi | 5 +-- weaviate/debug/types.py | 28 +++++++++++++++- 7 files changed, 132 insertions(+), 8 deletions(-) create mode 100644 mock_tests/test_debug.py diff --git a/integration/test_client_debug.py b/integration/test_client_debug.py index d9b2777c2..26449ba20 100644 --- a/integration/test_client_debug.py +++ b/integration/test_client_debug.py @@ -63,3 +63,20 @@ def test_get_object_multi_node( debug_obj = client.debug.get_object_over_rest(collection.name, uuid, node_name=node_name) assert debug_obj is not None assert str(debug_obj.uuid) == str(uuid) + + +def test_list_tasks(client_factory: ClientFactory) -> None: + client = client_factory() + + tasks = client.debug.list_tasks() + + assert isinstance(tasks, dict) + + +@pytest.mark.asyncio +async def test_list_tasks_async(async_client_factory: AsyncClientFactory) -> None: + client = await async_client_factory() + + tasks = await client.debug.list_tasks() + + assert isinstance(tasks, dict) diff --git a/mock_tests/test_debug.py b/mock_tests/test_debug.py new file mode 100644 index 000000000..eb2a18201 --- /dev/null +++ b/mock_tests/test_debug.py @@ -0,0 +1,56 @@ +import weaviate +from weaviate.classes.debug import DistributedTask +from pytest_httpserver import HTTPServer + + +def test_list_tasks(weaviate_client: weaviate.WeaviateClient, weaviate_mock: HTTPServer) -> None: + weaviate_mock.expect_request("/v1/tasks").respond_with_json( + { + "reindex": [ + { + "id": "task-1", + "version": 1, + "status": "running", + "startedAt": "2026-01-01T00:00:00Z", + "finishedNodes": ["node1"], + "payload": {"collection": "MyCollection"}, + }, + { + "id": "task-2", + "version": 1, + "status": "finished", + "startedAt": "2026-01-01T00:00:00Z", + "finishedAt": "2026-01-01T00:05:00Z", + "finishedNodes": ["node1", "node2"], + }, + ] + } + ) + + tasks = weaviate_client.debug.list_tasks() + + assert list(tasks.keys()) == ["reindex"] + assert len(tasks["reindex"]) == 2 + + first = tasks["reindex"][0] + assert isinstance(first, DistributedTask) + assert first.id == "task-1" + assert first.status == "running" + assert first.finished_at is None + assert first.finished_nodes == ["node1"] + assert first.payload == {"collection": "MyCollection"} + + second = tasks["reindex"][1] + assert second.id == "task-2" + assert second.status == "finished" + assert second.finished_nodes == ["node1", "node2"] + + +def test_list_tasks_empty( + weaviate_client: weaviate.WeaviateClient, weaviate_mock: HTTPServer +) -> None: + weaviate_mock.expect_request("/v1/tasks").respond_with_json({}) + + tasks = weaviate_client.debug.list_tasks() + + assert tasks == {} diff --git a/weaviate/classes/debug.py b/weaviate/classes/debug.py index 0f299d8d4..983fda02c 100644 --- a/weaviate/classes/debug.py +++ b/weaviate/classes/debug.py @@ -1,5 +1,7 @@ -from weaviate.debug.types import DebugRESTObject +from weaviate.debug.types import DebugRESTObject, DistributedTask, DistributedTaskUnit __all__ = [ "DebugRESTObject", + "DistributedTask", + "DistributedTaskUnit", ] diff --git a/weaviate/debug/async_.pyi b/weaviate/debug/async_.pyi index 7a1657ac8..58da41f52 100644 --- a/weaviate/debug/async_.pyi +++ b/weaviate/debug/async_.pyi @@ -1,8 +1,8 @@ -from typing import Optional +from typing import Dict, List, Optional from weaviate.classes.config import ConsistencyLevel from weaviate.connect.v4 import ConnectionAsync -from weaviate.debug.types import DebugRESTObject +from weaviate.debug.types import DebugRESTObject, DistributedTask from weaviate.types import UUID from .executor import _DebugExecutor @@ -17,3 +17,4 @@ class _DebugAsync(_DebugExecutor[ConnectionAsync]): node_name: Optional[str] = None, tenant: Optional[str] = None, ) -> Optional[DebugRESTObject]: ... + async def list_tasks(self) -> Dict[str, List[DistributedTask]]: ... diff --git a/weaviate/debug/executor.py b/weaviate/debug/executor.py index 818ead8a3..97c547ae2 100644 --- a/weaviate/debug/executor.py +++ b/weaviate/debug/executor.py @@ -1,11 +1,11 @@ -from typing import Dict, Generic, Optional +from typing import Dict, Generic, List, Optional from httpx import Response from weaviate.classes.config import ConsistencyLevel from weaviate.connect import executor from weaviate.connect.v4 import ConnectionType, _ExpectedStatusCodes -from weaviate.debug.types import DebugRESTObject +from weaviate.debug.types import DebugRESTObject, DistributedTask from weaviate.types import UUID @@ -50,3 +50,24 @@ def resp(response: Response) -> Optional[DebugRESTObject]: error_msg="Object was not retrieved", status_codes=_ExpectedStatusCodes(ok_in=[200, 404], error="get object"), ) + + def list_tasks(self) -> executor.Result[Dict[str, List[DistributedTask]]]: + """Use the REST API endpoint /tasks to list all distributed tasks currently active or available in the cluster. + + Distributed tasks are long-running background operations, such as reindexing, that are tracked + across the cluster's nodes. The returned mapping is keyed by task namespace. + """ + + def resp(response: Response) -> Dict[str, List[DistributedTask]]: + return { + namespace: [DistributedTask(**task) for task in tasks] + for namespace, tasks in response.json().items() + } + + return executor.execute( + response_callback=resp, + method=self._connection.get, + path="/tasks", + error_msg="Distributed tasks were not retrieved", + status_codes=_ExpectedStatusCodes(ok_in=200, error="list distributed tasks"), + ) diff --git a/weaviate/debug/sync.pyi b/weaviate/debug/sync.pyi index f0fce0263..038f271e4 100644 --- a/weaviate/debug/sync.pyi +++ b/weaviate/debug/sync.pyi @@ -1,8 +1,8 @@ -from typing import Optional +from typing import Dict, List, Optional from weaviate.classes.config import ConsistencyLevel from weaviate.connect.v4 import ConnectionSync -from weaviate.debug.types import DebugRESTObject +from weaviate.debug.types import DebugRESTObject, DistributedTask from weaviate.types import UUID from .executor import _DebugExecutor @@ -17,3 +17,4 @@ class _Debug(_DebugExecutor[ConnectionSync]): node_name: Optional[str] = None, tenant: Optional[str] = None, ) -> Optional[DebugRESTObject]: ... + def list_tasks(self) -> Dict[str, List[DistributedTask]]: ... diff --git a/weaviate/debug/types.py b/weaviate/debug/types.py index 88b11e08c..f81fec144 100644 --- a/weaviate/debug/types.py +++ b/weaviate/debug/types.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field @@ -15,3 +15,29 @@ class DebugRESTObject(BaseModel): uuid: uuid_package.UUID = Field(..., alias="id") vector: Optional[list[float]] = Field(None) vectors: Optional[Dict[str, list[float]]] = Field(None) + + +class DistributedTaskUnit(BaseModel): + """A unit of a distributed task.""" + + id: str = Field(...) # noqa: A003 + node_id: str = Field(..., alias="nodeId") + status: str = Field(...) + progress: Optional[float] = Field(None) + error: Optional[str] = Field(None) + updated_at: Optional[datetime] = Field(None, alias="updatedAt") + finished_at: Optional[datetime] = Field(None, alias="finishedAt") + + +class DistributedTask(BaseModel): + """Metadata about a distributed task running in the cluster.""" + + id: str = Field(...) # noqa: A003 + version: int = Field(...) + status: str = Field(...) + started_at: datetime = Field(..., alias="startedAt") + finished_at: Optional[datetime] = Field(None, alias="finishedAt") + finished_nodes: List[str] = Field(default_factory=list, alias="finishedNodes") + error: Optional[str] = Field(None) + payload: Optional[Dict[str, Any]] = Field(None) + units: Optional[List[DistributedTaskUnit]] = Field(None)