def get_task_cls()

in src/pydolphinscheduler/core/yaml_workflow.py [0:0]

• 7 lines of code
• 3 McCabe index (conditional complexity)

def get_task_cls(task_type) -> Task:
    """Get the task class object by task_type (case compatible)."""
    # only get task class from tasks.__all__
    all_task_types = {type_.capitalize(): type_ for type_ in tasks.__all__}
    task_type_cap = task_type.capitalize()
    if task_type_cap not in all_task_types:
        raise PyDSTaskNoFoundException("cant not find task %s" % task_type)

    standard_name = all_task_types[task_type_cap]
    return getattr(tasks, standard_name)