Merge pull request #598 from atlanhq/APP-5964

APP-5964 : Added a new param `workflow ID/name` to `client.workflow.monitor()`

Author: Aryamanz29

pyatlan/client/workflow.py

@@ -407,37 +407,48 @@ def update_owner(self, workflow_name: str, username: str) -> WorkflowResponse:
 
     @validate_arguments(config=dict(arbitrary_types_allowed=True))
     def monitor(
-        self, workflow_response: WorkflowResponse, logger: Optional[Logger] = None
+        self,
+        workflow_response: Optional[WorkflowResponse] = None,
+        logger: Optional[Logger] = None,
+        workflow_name: Optional[str] = None,
     ) -> Optional[AtlanWorkflowPhase]:
         """
         Monitor the status of the workflow's run.
 
         :param workflow_response: The workflow_response returned from running the workflow
         :param logger: the logger to log status information
-        (logging.INFO for summary info. logging:DEBUG for detail info)
+        (logging.INFO for summary info. logging.DEBUG for detail info)
+        :param workflow_name: name of the workflow to be monitored
         :returns: the status at completion or None if the workflow wasn't run
         :raises ValidationError: If the provided `workflow_response`, `logger` is invalid
         :raises AtlanError: on any API communication issue
         """
-        if workflow_response.metadata and workflow_response.metadata.name:
-            name = workflow_response.metadata.name
-            status: Optional[AtlanWorkflowPhase] = None
-            while status not in {
-                AtlanWorkflowPhase.SUCCESS,
-                AtlanWorkflowPhase.ERROR,
-                AtlanWorkflowPhase.FAILED,
-            }:
-                sleep(MONITOR_SLEEP_SECONDS)
-                if run_details := self._get_run_details(name):
-                    status = run_details.status
+        name = workflow_name or (
+            workflow_response.metadata.name
+            if workflow_response and workflow_response.metadata
+            else None
+        )
+
+        if not name:
+            if logger:
+                logger.info("Skipping workflow monitoring — nothing to monitor.")
+            return None
+
+        status: Optional[AtlanWorkflowPhase] = None
+        while status not in {
+            AtlanWorkflowPhase.SUCCESS,
+            AtlanWorkflowPhase.ERROR,
+            AtlanWorkflowPhase.FAILED,
+        }:
+            sleep(MONITOR_SLEEP_SECONDS)
+            if run_details := self._get_run_details(name):
+                status = run_details.status
                 if logger:
                     logger.debug("Workflow status: %s", status)
-            if logger:
-                logger.info("Workflow completion status: %s", status)
-            return status
+
         if logger:
-            logger.info("Skipping workflow monitoring — nothing to monitor.")
-        return None
+            logger.info("Workflow completion status: %s", status)
+        return status
 
     def _get_run_details(self, name: str) -> Optional[WorkflowSearchResult]:
         return self._find_latest_run(workflow_name=name)

tests/integration/test_workflow_client.py

@@ -157,6 +157,11 @@ def test_workflow_get_runs_and_stop(client: AtlanClient, workflow: WorkflowRespo
     workflow_status = client.workflow.monitor(workflow_response=workflow)
     assert workflow_status == AtlanWorkflowPhase.FAILED
 
+    # Test workflow monitoring by providing workflow name directly
+    workflow_name = workflow.metadata.name
+    workflow_status = client.workflow.monitor(workflow_name=workflow_name)
+    assert workflow_status == AtlanWorkflowPhase.FAILED
+
     # Test find run by id
     workflow_run = client.workflow.find_run_by_id(id=run.id)
     assert (

tests/unit/constants.py

@@ -527,7 +527,6 @@
             ],
             "instance of Logger expected",
         ),
-        ([None, "test-logger"], "none is not an allowed value"),
     ],
     "get_runs": [
         ([[123], AtlanWorkflowPhase.RUNNING, 123, 456], "str type expected"),