def get_latest_statement_name_for_execution_arn()

in lambda/python/rs_integration_function/ddb/ddb_state_table.py [0:0]

• 20 lines of code
• 3 McCabe index (conditional complexity)

    def get_latest_statement_name_for_execution_arn(cls, execution_arn: str) -> StatementName:
        response = ddb_state_table.query(
            KeyConditionExpression=Key(DDB_ID).eq(execution_arn),
            ProjectionExpression="#I",
            ExpressionAttributeNames={
                "#I": DDB_INVOCATION_ID
            },
            ConsistentRead=True,
            ReturnConsumedCapacity='NONE',
        )
        logger.debug({l_response: response})
        items = response['Items']
        if len(items) == 0:
            e = PreviousExecutionNotFound(f"No started statements found for {execution_arn}")
            logger.warning({l_exception: e, l_response: response}, stack_info=True)
            raise e
        return StatementName(
            execution_arn,
            invocation_id=str(max(float(i[DDB_INVOCATION_ID]) for i in items))
        )