def list_executions()

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

• 30 lines of code
• 6 McCabe index (conditional complexity)

    def list_executions(self, max_items=100, status_filter=None, html=False):
        """
        Lists the executions for the workflow.

        Args:
            max_items (int, optional): The maximum number of items to be returned. (default: 100)
            status_filter (ExecutionStatus, optional): If specified, only list the executions whose current status matches the given filter. (default: None)
            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:
            list(stepfunctions.workflow.Execution): List of workflow run instances.
        """
        if self.state_machine_arn is None:
            return ExecutionsList()

        logger.debug("Retrieving list of executions from AWS Step Functions.")
        paginator = self.client.get_paginator('list_executions')
        params = {
            'stateMachineArn': self.state_machine_arn,
            'PaginationConfig': {
                'MaxItems': max_items,
                'PageSize': 1000
            }
        }
        if status_filter is not None:
            params['statusFilter'] = status_filter.value
        response_iterator = paginator.paginate(**params)

        runs = [
            Execution(
                name=execution['name'],
                workflow=self,
                execution_arn=execution['executionArn'],
                start_date=execution['startDate'],
                stop_date=execution.get('stopDate', None),
                status=execution['status'],
                client=self.client
        ) for page in response_iterator for execution in page['executions']]
        executions_list = ExecutionsList(runs)

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