added the overloaded logic and integration test

Author: vaibhavatlan

pyatlan/client/workflow.py

@@ -29,6 +29,7 @@
     ReRunRequest,
     ScheduleQueriesSearchRequest,
     Workflow,
+    WorkflowMetadata,
     WorkflowResponse,
     WorkflowRunResponse,
     WorkflowSchedule,
@@ -407,20 +408,29 @@ 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,
+        workflow_name: Optional[str] = None,
+        logger: Optional[Logger] = None,
     ) -> Optional[AtlanWorkflowPhase]:
         """
         Monitor the status of the workflow's run.
 
         :param workflow_response: The workflow_response returned from running the workflow
+        :param workflow_name: name of the workflow to be monitored
         :param logger: the logger to log status information
         (logging.INFO for summary info. logging:DEBUG for detail info)
         :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
+        if workflow_name and not workflow_response:
+            workflow_response = WorkflowResponse(
+                metadata=WorkflowMetadata(name=workflow_name)
+            )
+
+        if workflow_response.metadata and workflow_response.metadata.name:  # type: ignore
+            name = workflow_response.metadata.name  # type: ignore
             status: Optional[AtlanWorkflowPhase] = None
             while status not in {
                 AtlanWorkflowPhase.SUCCESS,

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 (