Used in https://github.com/mhenc/airflow/blob/master/airflow/dag_processing/manager.py#L613

should be straighforward
subdag_task_groups = dag.task_group.get_task_group_dict()
        for group in subdag_task_groups.values():
            group.upstream_group_ids.intersection_update(subdag_task_groups)
            group.downstream_group_ids.intersection_update(subdag_task_groups)
            group.upstream_task_ids.intersection_update(dag.task_dict)
            group.downstream_task_ids.intersection_update(dag.task_dict)
        for t in dag.tasks:
            
            
            t.upstream_task_ids.intersection_update(dag.task_dict)
            t.downstream_task_ids.intersection_update(dag.task_dict)
        if len(dag.tasks) < len(self.tasks):
            dag.partial = True
        return dag
    def has_task(self, task_id: str):
        return task_id in self.task_dict
    def has_task_group(self, task_group_id: str) -> bool:
        return task_group_id in self.task_group_dict
    @cached_property
    def task_group_dict(self):
        return {k: v for k, v in self._task_group.get_task_group_dict().items() if k is not None}
    def get_task(self, task_id: str, include_subdags: bool = False) -> Operator:
        if task_id in self.task_dict:
            return self.task_dict[task_id]
        if include_subdags:
            for dag in self.subdags:
                if task_id in dag.task_dict:
                    return dag.task_dict[task_id]
        raise TaskNotFound(f"Task {task_id} not found")
    def pickle_info(self):