def list_events()

in src/stepfunctions/workflow/stepfunctions.py [0:0]

• 21 lines of code
• 4 McCabe index (conditional complexity)

    def list_events(self, max_items=100, reverse_order=False, html=False):
        """
        Lists the events in the workflow execution.

        Args:
            max_items (int, optional): The maximum number of items to be returned. (default: 100)
            reverse_order (bool, optional): Boolean flag set to `True` if the events should be listed in reverse chronological order. Set to `False`, if the order should be in chronological order. (default: False)
            html (bool, optional): Renders the list as an HTML table (If running in an IPython environment). If the parameter is not provided, or set to False, a Python list is returned. (default: False)

        Returns:
            dict: Object containing the list of workflow execution events. Refer to :meth:`.SFN.Client.get_execution_history()` for the response structure.
        """
        logger.debug("Retrieving list of history events for your execution from AWS Step Functions.")
        paginator = self.client.get_paginator('get_execution_history')
        params = {
            'executionArn': self.execution_arn,
            'reverseOrder': reverse_order,
            'PaginationConfig': {
                'MaxItems': max_items,
                'PageSize': 1000
            }
        }
        response_iterator = paginator.paginate(**params)

        events = []
        for page in response_iterator:
            for event in page['events']:
                events.append(event)
        events_list = EventsList(events)

        if html:
            return HTML(events_list.to_html())
        else:
            return events_list