Python Power Automate - Monitoring runs
Run history, failure reports, cancelling runs and resubmitting failed ones.
Every execution of a flow is a run, and runs are how you observe what your flows are actually doing - which succeeded, which failed and why. You create a Power Automate connection in the Dashboard, and the run history of every flow is available to your services.
Reading run history
conn.list_runs returns the run history of a flow, newest first. Each run's status and timing information are nested under its properties key.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetApprovalRunHistory(Service):
def handle(self):
# Get the connection by its Dashboard name
conn = self.microsoft.power_platform['My Power Automate']
# Read the flow's run history
response = conn.list_runs('flow-invoice-approval')
# Summarize each run
runs = []
for run in response['value']:
runs.append({
'id': run['name'],
'status': run['properties']['status'],
'started': run['properties']['startTime'],
})
self.response.payload = {'runs': runs}
Finding failed runs
A run's status is one of Succeeded, Failed, Running or Cancelled - filtering on it is how you build failure reports and alerts.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class ReportFailedApprovals(Service):
def handle(self):
conn = self.microsoft.power_platform['My Power Automate']
response = conn.list_runs('flow-invoice-approval')
# Collect the runs that failed
failed = []
for run in response['value']:
if run['properties']['status'] == 'Failed':
failed.append(run['name'])
# Let the on-call team know if there is anything to look at
if failed:
self.logger.warning('Failed approval runs: %s', failed)
self.response.payload = {'failed_count': len(failed), 'failed': failed}
Inspecting a single run
conn.get_run returns full details of one run, including its error information if it failed.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetRunDetails(Service):
input = 'flow_id', 'run_id'
def handle(self):
conn = self.microsoft.power_platform['My Power Automate']
run = conn.get_run(self.request.input.flow_id, self.request.input.run_id)
self.response.payload = {
'status': run['properties']['status'],
'started': run['properties']['startTime'],
}
Cancelling a run
Long-running or stuck runs can be cancelled - for instance, an approval that is no longer relevant.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class CancelStaleApproval(Service):
input = 'run_id'
def handle(self):
conn = self.microsoft.power_platform['My Power Automate']
conn.cancel_run('flow-invoice-approval', self.request.input.run_id)
self.response.payload = {'status': 'cancelled'}
Resubmitting a failed run
conn.resubmit_run executes the flow again with the same trigger data the original run had - the standard way to recover from transient failures.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class RetryFailedApprovals(Service):
def handle(self):
conn = self.microsoft.power_platform['My Power Automate']
response = conn.list_runs('flow-invoice-approval')
# Resubmit every failed run with its original trigger data
resubmitted = []
for run in response['value']:
if run['properties']['status'] == 'Failed':
conn.resubmit_run('flow-invoice-approval', run['name'])
resubmitted.append(run['name'])
self.response.payload = {'resubmitted': resubmitted}