This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
"""
        Adds the DAG into the bag, recurses into sub dags.
        :raises: AirflowDagCycleException if a cycle is detected in this dag or its subdags.
        :raises: AirflowDagDuplicatedIdException if this dag or its subdags already exists in the bag.
        """
        self._bag_dag(dag=dag, root_dag=root_dag, recursive=True)
    def _bag_dag(self, *, dag, root_dag, recursive):
        """Actual implementation of bagging a dag.
        The only purpose of this is to avoid exposing ``recursive`` in ``bag_dag()``,
        intended to only be used by the ``_bag_dag()`` implementation.
        """
        check_cycle(dag)  
        dag.resolve_template_files()
        dag.last_loaded = timezone.utcnow()
        try:
            
            settings.dag_policy(dag)
            for task in dag.tasks:
                settings.task_policy(task)
        except AirflowClusterPolicyViolation:
            raise
        except Exception as e:
            self.log.exception(e)
            raise AirflowClusterPolicyError(e)
        subdags = dag.subdags
        try: