Get details on SWF workflow failures

wff.py

import json
import time

import boto3


# aws swf list-closed-workflow-executions
# --domain $SWF_DOMAIN
# --type-filter name=$WORKFLOW_NAME
# --start-time-filter oldestDate=$(($(date +%s) - 24*60*60))
# | jq '.executionInfos|.[]|select(.closeStatus == "FAILED")
# |{workflow: .workflowType.name, closeStatus: .closeStatus,runId: .execution.runId, workflowId: .execution.workflowId}'

SWF_DOMAIN='prod-domain'
WORKFLOW_NAME='workflow-1'
FAILURES = []

def main():
    swf = boto3.client('swf')
    closed = swf.list_closed_workflow_executions(
        domain=WORKFLOW_DOMAIN',
        typeFilter=dict(name=WORKFLOW_NAME),
        startTimeFilter=dict(oldestDate=int(time.time()) - 7*24*60*60)
    )['executionInfos']
    # get failed
    failed_executions = [_execution for _execution in closed if
                         _execution['closeStatus'] == 'FAILED']
    for failed_execution in failed_executions:
        events = swf.get_workflow_execution_history(
            domain=SWF_DOMAIN,
            execution=dict(
                workflowId=failed_execution['execution']['workflowId'],
                runId=failed_execution['execution']['runId'],
            )
        )['events']
        failed_events = [_event['workflowExecutionFailedEventAttributes']
                         for _event in events
                         if _event.get('workflowExecutionFailedEventAttributes')]
        for _fe in failed_events:
            for _k, _v in _fe.items():
                if _k == 'details':
                    _fe['details'] = json.loads(_v)
        FAILURES.append(
            {'workflow': failed_execution['workflowType'],
             'execution': failed_execution['execution'],
             'failed_events': failed_events}
        )
    print(json.dumps(FAILURES, sort_keys=True, indent=2))



if __name__ == '__main__':
    main()