def list_inference_executions()

in getting_started/utils/lookout_equipment_utils.py [0:0]

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

    def list_inference_executions(self, 
                                  execution_status=None, 
                                  start_time=None, 
                                  end_time=None, 
                                  max_results=50):
        """
        This method lists all the past inference execution triggered by the
        current scheduler.
        
        PARAMS
        ======
            execution_status: string (default: None)
                Only keep the executions with a given status
                
            start_time: pandas.DateTime (default: None)
                Filters out the executions that happened before start_time
                
            end_time: pandas.DateTime (default: None)
                Filters out the executions that happened after end_time
                
            
            max_results: integer (default: 50)
                Max number of results you want to get out of this method
        
        RETURNS
        =======
            results_df: list of dict
                A list of all past inference executions, with each inference
                attributes stored in a python dictionary
        """
        # Built the execution request object:
        list_executions_request = {"MaxResults": max_results}
        list_executions_request["InferenceSchedulerName"] = self.scheduler_name
        if execution_status is not None:
            list_executions_request["Status"] = execution_status
        if start_time is not None:
            list_executions_request['DataStartTimeAfter'] = start_time
        if end_time is not None:
            list_executions_request['DataEndTimeBefore'] = end_time

        # Loops through all the inference executed by the current scheduler:
        has_more_records = True
        list_executions = []
        while has_more_records:
            list_executions_response = self.client.list_inference_executions(
                **list_executions_request
            )
            if "NextToken" in list_executions_response:
                list_executions_request["NextToken"] = list_executions_response["NextToken"]
            else:
                has_more_records = False

            list_executions = list_executions + \
                              list_executions_response["InferenceExecutionSummaries"]

        # Returns all the summaries in a list:
        self.execution_summaries = list_executions
        return list_executions