class_docstr: A collection of tasks.
  
  When set_downstream() or set_upstream() are called on the TaskGroup, it is applied across
  all tasks within the group if necessary.
  
  :param group_id: a unique, meaningful id for the TaskGroup. group_id must not conflict
      with group_id of TaskGroup or task_id of tasks in the DAG. Root TaskGroup has group_id
      set to None.
  :param prefix_group_id: If set to True, child task_id and group_id will be prefixed with
      this TaskGroup's group_id. If set to False, child task_id and group_id are not prefixed.
      Default is True.
  :param parent_group: The parent TaskGroup of this TaskGroup. parent_group is set to None
      for the root TaskGroup.
  :param dag: The DAG that this TaskGroup belongs to.
  :param default_args: A dictionary of default parameters to be used
      as constructor keyword parameters when initialising operators,
      will override default_args defined in the DAG level.
      Note that operators have the same hook, and precede those defined
      here, meaning that if your dict contains `'depends_on_past': True`
      here and `'depends_on_past': False` in the operator's call
      `default_args`, the actual value will be `False`.
  :param tooltip: The tooltip of the TaskGroup node when displayed in the UI
  :param ui_color: The fill color of the TaskGroup node when displayed in the UI
  :param ui_fgcolor: The label color of the TaskGroup node when displayed in the UI
  :param add_suffix_on_collision: If this task group name already exists,
      automatically add `__1` etc suffixes
class_docstr_tok: ["A","collection","of","tasks",".","When","set_downstream","(",")","or","set_upstream","(",")","are","called","on","the","TaskGroup",",","it","is","applied","across","all","tasks","within","the","group","if","necessary",".",":","param","group_id",":","a","unique",",","meaningful","id","for","the","TaskGroup",".","group_id","must","not","conflict","with","group_id","of","TaskGroup","or","task_id","of","tasks","in","the","DAG",".","Root","TaskGroup","has","group_id","set","to","None",".",":","param","prefix_group_id",":","If","set","to","True",",","child","task_id","and","group_id","will","be","prefixed","with","this","TaskGroup","'s","group_id",".","If","set","to","False",",","child","task_id","and","group_id","are","not","prefixed",".","Default","is","True",".",":","param","parent_group",":","The","parent","TaskGroup","of","this","TaskGroup",".","parent_group","is","set","to","None","for","the","root","TaskGroup",".",":","param","dag",":","The","DAG","that","this","TaskGroup","belongs","to",".",":","param","default_args",":","A","dictionary","of","default","parameters","to","be","used","as","constructor","keyword","parameters","when","initialising","operators",",","will","override","default_args","defined","in","the","DAG","level",".","Note","that","operators","have","the","same","hook",",","and","precede","those","defined","here",",","meaning","that","if","your","dict","contains","`","'depends_on_past","'",":","True","`","here","and","`","'depends_on_past","'",":","False","`","in","the","operator","'s","call","`","default_args","`",",","the","actual","value","will","be","`","False","`",".",":","param","tooltip",":","The","tooltip","of","the","TaskGroup","node","when","displayed","in","the","UI",":","param","ui_color",":","The","fill","color","of","the","TaskGroup","node","when","displayed","in","the","UI",":","param","ui_fgcolor",":","The","label","color","of","the","TaskGroup","node","when","displayed","in","the","UI",":","param","add_suffix_on_collision",":","If","this","task","group","name","already","exists",",","automatically","add","`","__1","`","etc","suffixes"]
code: def __init__(
      self,
      group_id: str | None,
      prefix_group_id: bool = True,
      parent_group: TaskGroup | None = None,
      dag: DAG | None = None,
      default_args: dict[str, Any] | None = None,
      tooltip: str = "",
      ui_color: str = "CornflowerBlue",
      ui_fgcolor: str = "#000",
      add_suffix_on_collision: bool = False,
  ):
      from airflow.models.dag import DagContext
  
      self.prefix_group_id = prefix_group_id
      self.default_args = copy.deepcopy(default_args or {})
  
      dag = dag or DagContext.get_current_dag()
  
      if group_id is None:
          # This creates a root TaskGroup.
          if parent_group:
              raise AirflowException(
                  "Root TaskGroup cannot have parent_group"
              )
          # used_group_ids is shared across all TaskGroups in the same DAG to keep track
          # of used group_id to avoid duplication.
          self.used_group_ids = set()
          self.dag = dag
      else:
          if prefix_group_id:
              # If group id is used as prefix, it should not contain spaces nor dots
              # because it is used as prefix in the task_id
              validate_group_key(group_id)
          else:
              if not isinstance(group_id, str):
                  raise ValueError("group_id must be str")
              if not group_id:
                  raise ValueError("group_id must not be empty")
  
          if not parent_group and not dag:
              raise AirflowException(
                  "TaskGroup can only be used inside a dag"
              )
  
          parent_group = (
              parent_group
              or TaskGroupContext.get_current_task_group(dag)
          )
          if not parent_group:
              raise AirflowException(
                  "TaskGroup must have a parent_group except for the root TaskGroup"
              )
          if dag is not parent_group.dag:
              raise RuntimeError(
                  "Cannot mix TaskGroups from different DAGs: %s and %s",
                  dag,
                  parent_group.dag,
              )
  
          self.used_group_ids = parent_group.used_group_ids
  
      # if given group_id already used assign suffix by incrementing largest used suffix integer
      # Example : task_group ==> task_group__1 -> task_group__2 -> task_group__3
      self._group_id = group_id
      self._check_for_group_id_collisions(add_suffix_on_collision)
  
      self.children: dict[str, DAGNode] = {}
  
      if parent_group:
          parent_group.add(self)
          self._update_default_args(parent_group)
  
      self.used_group_ids.add(self.group_id)
      if self.group_id:
          self.used_group_ids.add(self.downstream_join_id)
          self.used_group_ids.add(self.upstream_join_id)
  
      self.tooltip = tooltip
      self.ui_color = ui_color
      self.ui_fgcolor = ui_fgcolor
  
      # Keep track of TaskGroups or tasks that depend on this entire TaskGroup separately
      # so that we can optimize the number of edges when entire TaskGroups depend on each other.
      self.upstream_group_ids: set[str | None] = set()
      self.downstream_group_ids: set[str | None] = set()
      self.upstream_task_ids = set()
      self.downstream_task_ids = set()
  
imports: from __future__ import annotations
  import copy
  import functools
  import operator
  import weakref
  from typing import TYPE_CHECKING, Any, Generator, Iterator, Sequence
  import re2
  from airflow.compat.functools import cache
  from airflow.exceptions import AirflowDagCycleException, AirflowException, DuplicateTaskIdFound, TaskAlreadyInTaskGroup
  from airflow.models.taskmixin import DAGNode
  from airflow.serialization.enums import DagAttributeTypes
  from airflow.utils.helpers import validate_group_key
label_desc: Use image node_id 1 to create a new TaskGroup object from inherited base classes: DAGNode with example: obj = TaskGroup(group_id, prefix_group_id, parent_group, dag, default_args, tooltip, ui_color, ui_fgcolor, add_suffix_on_collision)