def list_inference_executions()

in apps/cloudwatch-dashboard/lambdas/scheduler-details/handler.py [0:0]

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

def list_inference_executions(scheduler_name,
                              execution_status=None, 
                              start_time=None, 
                              end_time=None, 
                              max_results=50):
    """
    This method lists all the past inference execution triggered by a
    given 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"] = 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 = l4e_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:
    return list_executions