I have list of executionArn of stepfunction execution, I am filtering those executionArn by status == 'FAILED'.
Now, how can I get the exact failure of the stepfunction in Python, (meaning: at which point the execution has failed)?
I have list of executionArn of stepfunction execution, I am filtering those executionArn by status == 'FAILED'.
Now, how can I get the exact failure of the stepfunction in Python, (meaning: at which point the execution has failed)?
You can do something like this:
import boto3
import jmespath
client = boto3.client('stepfunctions')
full_result = (
client
.get_paginator('get_execution_history')
.paginate(arn=<your arn>)
.build_full_result()
)
exn_failed_event = jmespath.search('events[?type==`ExecutionFailed`]', full_result)
The exn_failed_event dictionary will look something like this
[
{
'executionFailedEventDetails': {
'cause':
'An error occurred while executing '
"the state 'Lambda Invoke' (entered "
'at the event id #4). Invalid path '
"'$.foo' : No results for path: "
"$['foo']",
'error': 'States.Runtime'
},
'id': 11,
'previousEventId': 0,
'timestamp': datetime.datetime(2022, 7, 11, 19, 0, 24, 2000, tzinfo=tzlocal()),
'type': 'ExecutionFailed'
}
]
Based on this issue in boto3 repository, it looks like there is a search method that you can call on an iterator object.
So your actual code could be simplified — and gather less unneeded data — to:
import boto3
failed_events = boto3.client('stepfunctions')
.get_paginator('get_execution_history')
.paginate(arn=<your arn>)
.search('events[?type==`ExecutionFailed`]')