From 6ee6ba9726f139c3e88e06a282fa8a21624e22e5 Mon Sep 17 00:00:00 2001 From: muhammadshayan124 Date: Sun, 5 Jul 2026 16:30:49 +0500 Subject: [PATCH 1/2] 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) From dead9090036f65e8b11820caf43bb3c545fdb840 Mon Sep 17 00:00:00 2001 From: Muhammad Shayan Date: Mon, 13 Jul 2026 09:55:49 +0500 Subject: [PATCH 2/2] fix: skip list_tasks integration tests on servers without the endpoint The /v1/tasks REST endpoint landed on weaviate-core's stable/v1.37 branch on 2026-06-29, but this suite's CI-pinned Docker images for several matrix entries predate that merge and still respond 501 "not yet implemented". Skip rather than fail when that specific status code comes back, so the test stays meaningful once the pinned images catch up. --- integration/test_client_debug.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/integration/test_client_debug.py b/integration/test_client_debug.py index 26449ba20..c98ec7b2d 100644 --- a/integration/test_client_debug.py +++ b/integration/test_client_debug.py @@ -8,6 +8,12 @@ ) from weaviate.classes.config import DataType, Property from weaviate.classes.debug import DebugRESTObject +from weaviate.exceptions import UnexpectedStatusCodeError + +# The /v1/tasks REST endpoint landed on weaviate-core's stable/v1.37 branch on 2026-06-29 +# but isn't in every Docker image this suite's version matrix pins yet, so older builds +# still respond 501 "not yet implemented" rather than a real payload. +_DISTRIBUTED_TASKS_NOT_IMPLEMENTED = 501 def test_get_object_single_node( @@ -68,7 +74,12 @@ def test_get_object_multi_node( def test_list_tasks(client_factory: ClientFactory) -> None: client = client_factory() - tasks = client.debug.list_tasks() + try: + tasks = client.debug.list_tasks() + except UnexpectedStatusCodeError as e: + if e.status_code == _DISTRIBUTED_TASKS_NOT_IMPLEMENTED: + pytest.skip("distributed tasks endpoint not yet implemented on this server build") + raise assert isinstance(tasks, dict) @@ -77,6 +88,11 @@ def test_list_tasks(client_factory: ClientFactory) -> None: async def test_list_tasks_async(async_client_factory: AsyncClientFactory) -> None: client = await async_client_factory() - tasks = await client.debug.list_tasks() + try: + tasks = await client.debug.list_tasks() + except UnexpectedStatusCodeError as e: + if e.status_code == _DISTRIBUTED_TASKS_NOT_IMPLEMENTED: + pytest.skip("distributed tasks endpoint not yet implemented on this server build") + raise assert isinstance(tasks, dict)