Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions integration/test_client_debug.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
56 changes: 56 additions & 0 deletions mock_tests/test_debug.py
Original file line number Diff line number Diff line change
@@ -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 == {}
4 changes: 3 additions & 1 deletion weaviate/classes/debug.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from weaviate.debug.types import DebugRESTObject
from weaviate.debug.types import DebugRESTObject, DistributedTask, DistributedTaskUnit

__all__ = [
"DebugRESTObject",
"DistributedTask",
"DistributedTaskUnit",
]
5 changes: 3 additions & 2 deletions weaviate/debug/async_.pyi
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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]]: ...
25 changes: 23 additions & 2 deletions weaviate/debug/executor.py
Original file line number Diff line number Diff line change
@@ -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


Expand Down Expand Up @@ -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"),
)
5 changes: 3 additions & 2 deletions weaviate/debug/sync.pyi
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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]]: ...
28 changes: 27 additions & 1 deletion weaviate/debug/types.py
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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)
Loading