def _get_output_fetching_processes()

in airavata_django_portal_sdk/experiment_util/intermediate_output.py [0:0]

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

def _get_output_fetching_processes(experiment: ExperimentModel) -> List[ProcessModel]:
    "sort the processes (most recent first) and filter to just the output fetching ones"
    processes: List[ProcessModel] = sorted(experiment.processes, key=lambda p: p.creationTime, reverse=True) if experiment.processes else []
    output_fetching_processes: List[ProcessModel] = []
    for process in processes:
        if any(map(lambda t: t.taskType == TaskTypes.OUTPUT_FETCHING, process.tasks)):
            output_fetching_processes.append(process)
    return output_fetching_processes