How to get status of workflow in other workflow? (incl. handling forced terminations)

Community Support
2

You can just use the Temporal client that you pass in when creating the activity class just before registering, e.g.

class Activities:
    def __init__(self, client: temporalio.client.Client) -> None:
        self._client = client

    @activity.defn
    async def is_workflow_running(self, workflow_id: str) -> bool:
        # May want to catch not-found RPCError here if it's not required to exist
        desc = await self._client.get_workflow_handle(workflow_id).describe()
        return desc.status == temporalio.client.WorkflowExecutionStatus.RUNNING

Correct, there is no workflow-side way for a workflow to reliably report that it has completed, because some completions occur server side (timeouts and termination). So you have to poll for status. If you needed to check the status of multiple you could use the listing feature on the client.

Also, since you’re polling regularly, consider using the infrequent polling pattern here to avoid unnecessarily increasing up history waiting for workflow complete. (so you’d call the activity wait_for_workflow_complete and raise an error while still running with fixed backoff)

1 Like