def get_unique_tasks()

in source/generate_dag.py [0:0]

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

def get_unique_tasks(config_data):
    """
    Extracts unique task IDs from a dictionary containing task and task group definitions.

    Args:
      config_data: A dictionary containing task and task group definitions.

    Returns:
      A list of unique task IDs, including task group IDs.
    """
    all_task_ids = []
    
    # Extract task IDs from regular tasks
    for task in config_data.get('tasks', []):
        all_task_ids.append(task['task_id'])

    # Extract task IDs from task groups and their tasks
    for group in config_data.get('task_groups', []):
        all_task_ids.append(group['group_id'])  # Add the group_id as a task
        if 'tasks' in group:
            for task in group['tasks']:
                all_task_ids.append(task['task_id'])

    # Return unique task IDs
    return sorted(list(set(all_task_ids)))