feat(platform): add assign and complete to TasksService

packages/uipath-platform/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath-platform"
-version = "0.2.41"
+version = "0.2.42"
 description = "HTTP client library for programmatic access to UiPath Platform"
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"

packages/uipath-platform/src/uipath/platform/action_center/_tasks_service.py

@@ -20,6 +20,7 @@
 from ..common._execution_context import UiPathExecutionContext
 from ..common._folder_context import FolderContext, header_folder
 from ..common._models import Endpoint, RequestSpec
+from ..errors import OperationFailedException
 from .task_schema import TaskSchema
 from .tasks import Task, TaskRecipient, TaskRecipientType
 
@@ -327,6 +328,62 @@ def _retrieve_action_spec(
     )
 
 
+_COMPLETE_ENDPOINTS = {
+    "AppTask": "/orchestrator_/tasks/AppTasks/CompleteAppTask",
+    "FormTask": "/orchestrator_/forms/TaskForms/CompleteTask",
+}
+
+
+def _assign_by_user_spec(
+    task_id: int, user_id: Optional[int], user_name_or_email: Optional[str]
+) -> RequestSpec:
+    if (user_id is None) == (user_name_or_email is None):
+        raise ValueError("Provide exactly one of user_id or user_name_or_email")
+    assignment: Dict[str, Any] = {"taskId": task_id}
+    if user_id is not None:
+        assignment["userId"] = user_id
+    else:
+        assignment["userNameOrEmail"] = user_name_or_email
+    return RequestSpec(
+        method="POST",
+        endpoint=Endpoint(
+            "/orchestrator_/odata/Tasks/UiPath.Server.Configuration.OData.AssignTasks"
+        ),
+        json={"taskAssignments": [assignment]},
+    )
+
+
+def _raise_on_assignment_errors(task_id: int, body: Dict[str, Any]) -> None:
+    # AssignTasks answers 200 and lists per-task failures in "value".
+    for error in body.get("value") or []:
+        raise OperationFailedException(
+            operation_id=str(error.get("TaskId", task_id)),
+            status=str(error.get("ErrorCode")),
+            error=str(error.get("ErrorMessage")),
+            operation_name="Task assignment",
+        )
+
+
+def _complete_spec(
+    task_id: int,
+    action: Optional[str],
+    data: Optional[Dict[str, Any]],
+    task_type: str,
+    folder_key: Optional[str],
+    folder_path: Optional[str],
+) -> RequestSpec:
+    return RequestSpec(
+        method="POST",
+        endpoint=Endpoint(
+            _COMPLETE_ENDPOINTS.get(
+                task_type, "/orchestrator_/tasks/GenericTasks/CompleteTask"
+            )
+        ),
+        json={"taskId": task_id, "action": action, "data": data or {}},
+        headers=header_folder(folder_key, folder_path),
+    )
+
+
 async def _assign_task_spec(
     self, task_key: str, assignee: str | None, task_recipient: TaskRecipient | None
 ) -> RequestSpec:
@@ -902,6 +959,106 @@ async def retrieve_async(
 
         return Task.model_validate(response.json())
 
+    @traced(name="tasks_assign", run_type="uipath")
+    def assign(
+        self,
+        task_id: int,
+        *,
+        user_id: Optional[int] = None,
+        user_name_or_email: Optional[str] = None,
+    ) -> None:
+        """Assigns a task to a single user synchronously.
+
+        Args:
+            task_id: The id of the task to assign
+            user_id: The Orchestrator id of the assignee
+            user_name_or_email: The user name or email of the assignee
+
+        Raises:
+            ValueError: If not exactly one of user_id or user_name_or_email is given
+            OperationFailedException: If Action Center rejects the assignment
+        """
+        spec = _assign_by_user_spec(task_id, user_id, user_name_or_email)
+        response = self.request(spec.method, spec.endpoint, json=spec.json)
+        _raise_on_assignment_errors(task_id, response.json())
+
+    @traced(name="tasks_assign", run_type="uipath")
+    async def assign_async(
+        self,
+        task_id: int,
+        *,
+        user_id: Optional[int] = None,
+        user_name_or_email: Optional[str] = None,
+    ) -> None:
+        """Assigns a task to a single user asynchronously.
+
+        Args:
+            task_id: The id of the task to assign
+            user_id: The Orchestrator id of the assignee
+            user_name_or_email: The user name or email of the assignee
+
+        Raises:
+            ValueError: If not exactly one of user_id or user_name_or_email is given
+            OperationFailedException: If Action Center rejects the assignment
+        """
+        spec = _assign_by_user_spec(task_id, user_id, user_name_or_email)
+        response = await self.request_async(spec.method, spec.endpoint, json=spec.json)
+        _raise_on_assignment_errors(task_id, response.json())
+
+    @traced(name="tasks_complete", run_type="uipath")
+    def complete(
+        self,
+        task_id: int,
+        action: Optional[str] = None,
+        data: Optional[Dict[str, Any]] = None,
+        *,
+        task_type: str = "AppTask",
+        folder_path: Optional[str] = None,
+        folder_key: Optional[str] = None,
+    ) -> None:
+        """Completes a task synchronously, which resumes a job waiting on it.
+
+        Only the current assignee can complete a task, so assign it first.
+
+        Args:
+            task_id: The id of the task to complete
+            action: The outcome to submit, for example "Approve"
+            data: The task data to submit
+            task_type: "AppTask", "FormTask", or any other task type
+            folder_path: Optional folder path of the task
+            folder_key: Optional folder key of the task
+        """
+        spec = _complete_spec(task_id, action, data, task_type, folder_key, folder_path)
+        self.request(spec.method, spec.endpoint, json=spec.json, headers=spec.headers)
+
+    @traced(name="tasks_complete", run_type="uipath")
+    async def complete_async(
+        self,
+        task_id: int,
+        action: Optional[str] = None,
+        data: Optional[Dict[str, Any]] = None,
+        *,
+        task_type: str = "AppTask",
+        folder_path: Optional[str] = None,
+        folder_key: Optional[str] = None,
+    ) -> None:
+        """Completes a task asynchronously, which resumes a job waiting on it.
+
+        Only the current assignee can complete a task, so assign it first.
+
+        Args:
+            task_id: The id of the task to complete
+            action: The outcome to submit, for example "Approve"
+            data: The task data to submit
+            task_type: "AppTask", "FormTask", or any other task type
+            folder_path: Optional folder path of the task
+            folder_key: Optional folder key of the task
+        """
+        spec = _complete_spec(task_id, action, data, task_type, folder_key, folder_path)
+        await self.request_async(
+            spec.method, spec.endpoint, json=spec.json, headers=spec.headers
+        )
+
     async def _get_app_key_and_schema_async(
         self,
         app_name: str | None,

packages/uipath-platform/tests/services/test_actions_service.py

@@ -8,7 +8,8 @@
 from uipath.platform.action_center import Task
 from uipath.platform.action_center._tasks_service import TasksService
 from uipath.platform.action_center.tasks import TaskRecipient, TaskRecipientType
-from uipath.platform.constants import HEADER_USER_AGENT
+from uipath.platform.constants import HEADER_FOLDER_KEY, HEADER_USER_AGENT
+from uipath.platform.errors import OperationFailedException
 
 
 @pytest.fixture
@@ -1109,3 +1110,124 @@ def test_create_task_source_metadata_overrides_a_built_in_key_on_collision(
     # jit_debug_env sets UIPATH_TRACE_ID=trace-1, which _apply_task_source uses to
     # build the built-in InstanceId; the caller-supplied key wins on collision.
     assert task_source_metadata["InstanceId"] == "caller-supplied-instance-id"
+
+
+ASSIGN_TASKS = (
+    "/orchestrator_/odata/Tasks/UiPath.Server.Configuration.OData.AssignTasks"
+)
+
+
+@pytest.fixture
+def api(base_url: str, org: str, tenant: str) -> str:
+    return f"{base_url}{org}{tenant}"
+
+
+def test_assign_to_a_user_id(
+    httpx_mock: HTTPXMock, service: TasksService, api: str
+) -> None:
+    httpx_mock.add_response(url=f"{api}{ASSIGN_TASKS}", json={"value": []})
+
+    service.assign(555, user_id=42)
+
+    request = httpx_mock.get_request()
+    assert request is not None
+    assert json.loads(request.content) == {
+        "taskAssignments": [{"taskId": 555, "userId": 42}]
+    }
+
+
+async def test_assign_async_to_a_user_name(
+    httpx_mock: HTTPXMock, service: TasksService, api: str
+) -> None:
+    httpx_mock.add_response(url=f"{api}{ASSIGN_TASKS}", json={"value": []})
+
+    await service.assign_async(555, user_name_or_email="reviewer@example.com")
+
+    request = httpx_mock.get_request()
+    assert request is not None
+    assert json.loads(request.content) == {
+        "taskAssignments": [{"taskId": 555, "userNameOrEmail": "reviewer@example.com"}]
+    }
+
+
+async def test_assign_async_raises_on_a_failure_reported_under_http_200(
+    httpx_mock: HTTPXMock, service: TasksService, api: str
+) -> None:
+    httpx_mock.add_response(
+        url=f"{api}{ASSIGN_TASKS}",
+        json={
+            "value": [
+                {
+                    "TaskId": 555,
+                    "ErrorCode": 1002,
+                    "ErrorMessage": "Action does not exist",
+                }
+            ]
+        },
+    )
+
+    with pytest.raises(OperationFailedException, match="Action does not exist"):
+        await service.assign_async(555, user_id=42)
+
+
+@pytest.mark.parametrize(
+    "kwargs", [{}, {"user_id": 42, "user_name_or_email": "reviewer@example.com"}]
+)
+def test_assign_needs_exactly_one_assignee(
+    service: TasksService, kwargs: dict[str, Any]
+) -> None:
+    with pytest.raises(ValueError, match="exactly one"):
+        service.assign(555, **kwargs)
+
+
+@pytest.mark.parametrize(
+    ("task_type", "endpoint"),
+    [
+        ("AppTask", "/orchestrator_/tasks/AppTasks/CompleteAppTask"),
+        ("FormTask", "/orchestrator_/forms/TaskForms/CompleteTask"),
+        ("ExternalTask", "/orchestrator_/tasks/GenericTasks/CompleteTask"),
+    ],
+)
+async def test_complete_async_posts_to_the_endpoint_of_the_task_type(
+    httpx_mock: HTTPXMock,
+    service: TasksService,
+    api: str,
+    task_type: str,
+    endpoint: str,
+) -> None:
+    httpx_mock.add_response(url=f"{api}{endpoint}", status_code=204)
+
+    await service.complete_async(
+        555,
+        "Approve",
+        {"comment": "ok"},
+        task_type=task_type,
+        folder_key="folder-key",
+    )
+
+    request = httpx_mock.get_request()
+    assert request is not None
+    assert request.headers[HEADER_FOLDER_KEY] == "folder-key"
+    assert json.loads(request.content) == {
+        "taskId": 555,
+        "action": "Approve",
+        "data": {"comment": "ok"},
+    }
+
+
+def test_complete_an_app_task(
+    httpx_mock: HTTPXMock, service: TasksService, api: str
+) -> None:
+    httpx_mock.add_response(
+        url=f"{api}/orchestrator_/tasks/AppTasks/CompleteAppTask", status_code=204
+    )
+
+    service.complete(555, "Approve")
+
+    request = httpx_mock.get_request()
+    assert request is not None
+    assert json.loads(request.content) == {
+        "taskId": 555,
+        "action": "Approve",
+        "data": {},
+    }