def build_scheduler_details()

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

• 41 lines of code
• 2 McCabe index (conditional complexity)

def build_scheduler_details(scheduler_name):
    """
    Extract the scheduler attributes from the scheduler name passed as an argument
    
    Params:
        scheduler_name (string): name of the scheduler to get the attributes for
    """
    # Get attributes from the scheduler
    response      = l4e_client.describe_inference_scheduler(InferenceSchedulerName=scheduler_name)
    input_config  = response['DataInputConfiguration']
    output_config = response['DataOutputConfiguration']
    
    tz_offset        = input_config['InputTimeZoneOffset']
    input_bucket     = input_config['S3InputConfiguration']['Bucket']
    input_prefix     = input_config['S3InputConfiguration']['Prefix']
    input_s3_path    = f's3://{input_bucket}/{input_prefix}'
    timestamp_format = input_config['InferenceInputNameConfiguration']['TimestampFormat']
    delimiter        = input_config['InferenceInputNameConfiguration']['ComponentTimestampDelimiter']
    output_bucket    = output_config['S3OutputConfiguration']['Bucket']
    output_prefix    = output_config['S3OutputConfiguration']['Prefix']
    output_s3_path   = f's3://{output_bucket}/{output_prefix}'
    frequency        = response['DataUploadFrequency']
    
    date_format = '%Y-%m-%d %H:%M:%S'
    current_time, start_time, end_time, next_wakeup_time, next_timestamp = get_next_time_range(timestamp_format, frequency)
    
    next_execution = '<ul>'
    next_execution += '<li>Current time is <b>' + datetime.strftime(current_time, date_format) + '</b></li>'
    next_execution += '<li>Next execution time: <b>' + datetime.strftime(next_wakeup_time, date_format) + '</b></li>'
    next_execution += '<li>Next file: ' + f'<b><i>&lt;component&gt;</i>{delimiter}{next_timestamp}.csv</b>' + '</li>'
    next_execution += '<li>' + f'Timestamps must be between <b>{start_time}</b> and <b>{end_time}</b>' + '</li>'
    next_execution += '</ul>'
    
    num_executions, last_execution_time, last_success_time = get_last_execution(scheduler_name, date_format)
    last_execution = '<ul>'
    last_execution += f'<li>Executed <b>{num_executions}</b> times</li>'
    last_execution += f'<li>Last execution time: <b>{last_execution_time}</b></li>'
    
    if last_execution_time == last_success_time:
        last_execution += f'<li>Last successful execution: <b>{last_success_time}</b></li>'
    else:
        last_execution += f'<li>Last successful execution: <b><span style="color: #CC0000">{last_success_time}</span></b></li>'
    last_execution += '</ul>'
    
    # Build a dictionnary with all the model 
    # parameters we want to expose in the widget:
    scheduler_infos = dict()
    scheduler_infos.update({
        'Input': input_s3_path,
        'Output': output_s3_path,
        'File format': f'<i>&lt;component&gt;</i>{delimiter}{timestamp_format}.csv',
        'Next execution': next_execution,
        'Last execution': last_execution
    })

    # Generates the HTML of the widget:
    html = scheduler_info_widget(scheduler_infos)
        
    return html