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

should be straighforward
existing_task_outlet_refs_dict = collections.defaultdict(set)
        for dag_id, orm_dag in existing_dags.items():
            for todr in orm_dag.task_outlet_dataset_references:
                existing_task_outlet_refs_dict[(dag_id, todr.task_id)].add(todr)
        
        for (dag_id, task_id), uri_list in outlet_references.items():
            task_refs_needed = {
                TaskOutletDatasetReference(dataset_id=stored_datasets[uri].id, dag_id=dag_id, task_id=task_id)
                for uri in uri_list
            }
            task_refs_stored = existing_task_outlet_refs_dict[(dag_id, task_id)]
            task_refs_to_add = {x for x in task_refs_needed if x not in task_refs_stored}
            session.bulk_save_objects(task_refs_to_add)
            for obj in task_refs_stored - task_refs_needed:
                session.delete(obj)
        
        
        session.flush()
        for dag in dags:
            cls.bulk_write_to_db(dag.subdags, processor_subdir=processor_subdir, session=session)
    @provide_session
    def sync_to_db(self, processor_subdir: str | None = None, session=NEW_SESSION):
        """
        Save attributes about this DAG to the DB. Note that this method
        can be called for both DAGs and SubDAGs. A SubDag is actually a
        SubDagOperator.
        :return: None
        """
        self.bulk_write_to_db([self], processor_subdir=processor_subdir, session=session)
    def get_default_view(self):