def validate_create_task_dependency()

in source/generate_dag.py [0:0]

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

def validate_create_task_dependency(yaml_config: dict) -> dict:
    """Validate and create task dependency for the DAG.

    Args:
        yaml_config (dict): YAML configuration file for the task.

    Returns:
        dict: task_dependency_type (custom or default) to create task dependency.

    Raises:
        ValueError: If default_task_dependency is not boolean or 
                    if custom_task_dependency doesn't match config tasks.
    """
    try:
        default_dependency = yaml_config['task_dependency']['default_task_dependency']
        if not isinstance(default_dependency, bool):  # More concise type check
            raise ValueError("Invalid default_task_dependency value. Acceptable values are True or False")
    except KeyError as e:
        raise ValueError(f"Missing key in yaml_config: {e}") from e  # More informative error message

    task_list = get_unique_tasks(yaml_config)

    if default_dependency:
        return {"task_dependency_type": "default"}

    try:
        custom_dependency = yaml_config['task_dependency']['custom_task_dependency']
    except KeyError as e:
        raise ValueError(f"Missing key in yaml_config: {e}") from e

    print("Task Validation: validating tasks for custom dependency")

    task_dependency = []
    custom_tasks = set()
    for dependency_chain in custom_dependency:
        dependency_chain = dependency_chain.strip('"')
        task_dependency.append(dependency_chain)
        custom_tasks.update(re.findall(r'[\w_]+', dependency_chain))

    if sorted(custom_tasks) != task_list:
        print(f"List of config tasks: {task_list}")
        print(f"List of custom tasks: {sorted(custom_tasks)}")  # Sort for consistent comparison
        raise ValueError("Validation error: Tasks in custom_task_dependency don't match config tasks")

    print("Task Validation: task validation successful")
    return {"task_dependency_type": "custom", "task_dependency": task_dependency}