feat: Execute work item API

docs/api_reference/work_item.rst

@@ -13,6 +13,7 @@ nisystemlink.clients.work_item
    .. automethod:: update_work_items
    .. automethod:: schedule_work_items
    .. automethod:: delete_work_items
+   .. automethod:: execute_work_item
    .. automethod:: create_work_item_templates
    .. automethod:: query_work_item_templates
    .. automethod:: update_work_item_templates

docs/getting_started.rst

@@ -456,13 +456,13 @@ default connection. The default connection depends on your environment.
 
 With a :class:`.WorkItemClient` object, you can:
 
-* Create, query, get, update, schedule and delete work items
+* Create, query, get, update, schedule, delete and execute work items
 * Create, query, update and delete work item templates
 
 Examples
 ~~~~~~~~
 
-Create, query, get, update, schedule and delete work items
+Create, query, get, update, schedule, delete and execute work items
 
 .. literalinclude:: ../examples/work_item/work_items.py
    :language: python

examples/work_item/work_items.py

@@ -105,6 +105,7 @@
         dashboard=Dashboard(
             id="DashboardId", variables={"product": "PXIe-4080", "location": "Lab1"}
         ),
+        workflow_id="example-workflow-id",
         execution_actions=[
             ManualExecution(action="boot", type="MANUAL"),
             JobExecution(
@@ -202,6 +203,39 @@
             f"Scheduled work item with ID: {schedule_work_items_response.scheduled_work_items[0].id}"
         )
 
+# Execute work item action
+if created_work_item_id is not None:
+    try:
+        execute_response = client.execute_work_item(
+            work_item_id=created_work_item_id, action="START"
+        )
+        if execute_response.error is not None:
+            print(f"Execution failed: {execute_response.error.message}")
+        elif execute_response.result is not None:
+            print(f"Executed action successfully. Type: {execute_response.result.type}")
+
+            # Check for result-level error
+            if execute_response.result.error is not None:
+                print(f"Execution error: {execute_response.result.error.message}")
+            else:
+                # Handle type-specific fields
+                if execute_response.result.type == "NOTEBOOK":
+                    if execute_response.result.execution_id:
+                        print(
+                            f"Notebook execution ID: {execute_response.result.execution_id}"
+                        )
+                elif execute_response.result.type == "JOB":
+                    if execute_response.result.job_ids:
+                        print(f"Job IDs: {', '.join(execute_response.result.job_ids)}")
+                elif execute_response.result.type == "MANUAL":
+                    print("Manual execution completed")
+                elif execute_response.result.type == "SCHEDULE":
+                    print("Work item scheduled")
+                elif execute_response.result.type == "UNSCHEDULE":
+                    print("Work item unscheduled")
+    except Exception as e:
+        print(f"Could not execute action: {e}")
+
 # Delete work item
 if created_work_item_id is not None:
     client.delete_work_items(ids=[created_work_item_id])

nisystemlink/clients/work_item/_work_item_client.py

@@ -5,7 +5,7 @@
 from nisystemlink.clients.core._uplink._base_client import BaseClient
 from nisystemlink.clients.core._uplink._methods import get, post
 from nisystemlink.clients.work_item import models
-from uplink import Field, retry
+from uplink import Field, Path, retry
 
 
 @retry(
@@ -132,6 +132,27 @@ def delete_work_items(
         """
         ...
 
+    @post(
+        "workitems/{workItemId}/execute",
+        args=[Path(name="workItemId"), Field("action")],
+    )
+    def execute_work_item(
+        self, work_item_id: str, action: str
+    ) -> models.ExecuteWorkItemResponse:
+        """Executes the specified action for the work item.
+
+        Args:
+            work_item_id: The ID of the work item the action will be performed on.
+            action: The action to execute on the work item.
+
+        Returns:
+            The response containing the execution result or error information.
+
+        Raises:
+            ApiException: if unable to communicate with the `/niworkitem` service or provided invalid arguments.
+        """
+        ...
+
     @post("workitem-templates", args=[Field("workItemTemplates")])
     def create_work_item_templates(
         self, work_item_templates: List[models.CreateWorkItemTemplateRequest]

nisystemlink/clients/work_item/models/__init__.py

@@ -58,6 +58,11 @@
 )
 from ._update_work_items_request import UpdateWorkItemsRequest
 
+from ._execute_work_item_response import (
+    ExecuteWorkItemResponse,
+    ExecutionResult,
+)
+
 from ._create_work_item_template_request import CreateWorkItemTemplateRequest
 from ._create_work_item_templates_partial_success_response import (
     CreateWorkItemTemplatesPartialSuccessResponse,

nisystemlink/clients/work_item/models/_execute_work_item_response.py

@@ -0,0 +1,30 @@
+from typing import List, Literal
+
+from nisystemlink.clients.core._api_error import ApiError
+from nisystemlink.clients.core._uplink._json_model import JsonModel
+
+
+class ExecutionResult(JsonModel):
+    """Result of executing a work item action."""
+
+    type: Literal["NONE", "MANUAL", "NOTEBOOK", "JOB", "SCHEDULE", "UNSCHEDULE"]
+    """Type of execution."""
+
+    error: ApiError | None = None
+    """Error information if the execution encountered an error."""
+
+    execution_id: str | None = None
+    """The notebook execution ID. Only populated when type is NOTEBOOK."""
+
+    job_ids: List[str] | None = None
+    """The list of job IDs. Only populated when type is JOB."""
+
+
+class ExecuteWorkItemResponse(JsonModel):
+    """Response for executing a work item action."""
+
+    error: ApiError | None = None
+    """Error information if the action failed."""
+
+    result: ExecutionResult | None = None
+    """Result of the action execution."""

tests/integration/work_item/test_work_item_client.py

@@ -3,6 +3,8 @@
 from typing import List
 
 import pytest
+import responses
+from nisystemlink.clients.core import ApiException
 from nisystemlink.clients.core._http_configuration import HttpConfiguration
 from nisystemlink.clients.work_item import WorkItemClient
 from nisystemlink.clients.work_item.models import (
@@ -41,6 +43,8 @@
     WorkItemTemplateField,
 )
 
+BASE_URL = "https://test-api.lifecyclesolutions.ni.com"
+
 
 @pytest.fixture(scope="class")
 def client(enterprise_config: HttpConfiguration) -> WorkItemClient:
@@ -420,6 +424,119 @@ def test__delete_work_item(self, client: WorkItemClient, create_work_items):
         )
         assert len(query_deleted_work_item_response.work_items) == 0
 
+    @pytest.mark.parametrize(
+        "execution_type,action,extra_fields,expected_values",
+        [
+            ("MANUAL", "START", {}, {}),
+            (
+                "NOTEBOOK",
+                "RUN_NOTEBOOK",
+                {"execution_id": "notebook-execution-123"},
+                {"execution_id": "notebook-execution-123"},
+            ),
+            (
+                "JOB",
+                "RUN_JOBS",
+                {"job_ids": ["job-1", "job-2", "job-3"]},
+                {"job_ids": ["job-1", "job-2", "job-3"]},
+            ),
+            ("SCHEDULE", "SCHEDULE", {}, {}),
+            ("UNSCHEDULE", "UNSCHEDULE", {}, {}),
+        ],
+    )
+    @responses.activate
+    def test__execute_work_item_with_action__returns_execution_result(
+        self,
+        client: WorkItemClient,
+        execution_type: str,
+        action: str,
+        extra_fields: dict,
+        expected_values: dict,
+    ):
+        work_item_id = "test-work-item-id"
+
+        return_value = {
+            "result": {
+                "type": execution_type,
+                "error": None,
+                **extra_fields,
+            },
+            "error": None,
+        }
+
+        responses.add(
+            responses.POST,
+            f"{BASE_URL}/niworkitem/v1/workitems/{work_item_id}/execute",
+            json=return_value,
+            status=200,
+        )
+
+        execute_response = client.execute_work_item(
+            work_item_id=work_item_id, action=action
+        )
+
+        assert execute_response is not None
+        assert execute_response.error is None
+        assert execute_response.result is not None
+        assert execute_response.result.type == execution_type
+        assert execute_response.result.error is None
+
+        # Verify type-specific fields
+        for field, expected_value in expected_values.items():
+            assert getattr(execute_response.result, field) == expected_value
+
+    def test__execute_work_item_with_invalid_id__raises_ApiException_NotFound(
+        self, client: WorkItemClient
+    ):
+        invalid_work_item_id = "invalid-work-item-id"
+
+        with pytest.raises(ApiException) as exception_info:
+            client.execute_work_item(work_item_id=invalid_work_item_id, action="START")
+
+        assert exception_info.value.http_status_code == 404
+
+    @responses.activate
+    def test__execute_work_item_with_result_level_error__returns_result_with_error(
+        self, client: WorkItemClient
+    ):
+        work_item_id = "test-work-item-id"
+
+        return_value = {
+            "result": {
+                "type": "NOTEBOOK",
+                "execution_id": None,
+                "error": {
+                    "name": "Skyline.ExecutionFailed",
+                    "code": -251050,
+                    "message": "Notebook execution failed due to invalid parameters.",
+                    "args": [],
+                    "innerErrors": [],
+                },
+            },
+            "error": None,
+        }
+
+        responses.add(
+            responses.POST,
+            f"{BASE_URL}/niworkitem/v1/workitems/{work_item_id}/execute",
+            json=return_value,
+            status=200,
+        )
+
+        execute_response = client.execute_work_item(
+            work_item_id=work_item_id, action="RUN_NOTEBOOK"
+        )
+
+        assert execute_response is not None
+        assert execute_response.error is None
+        assert execute_response.result is not None
+        assert execute_response.result.type == "NOTEBOOK"
+        assert execute_response.result.execution_id is None
+        assert execute_response.result.error is not None
+        assert execute_response.result.error.code == -251050
+        assert execute_response.result.error.message is not None
+        assert "execution failed" in execute_response.result.error.message
+
     def test__create_work_item_template__returns_created_work_item_template(
         self, create_work_item_templates
     ):