return cls.bulk_write_to_db(dags=dags, session=session)
    @classmethod
    @provide_session
    def bulk_write_to_db(
        cls,
        dags: Collection[DAG],
        processor_subdir: str | None = None,
        session=NEW_SESSION,
    ):
        """
        Ensure the DagModel rows for the given dags are up-to-date in the dag table in the DB, including
        calculated fields.
        Note that this method can be called for both DAGs and SubDAGs. A SubDag is actually a SubDagOperator.
        :param dags: the DAG objects to save to the DB
        :return: None
        """
        if not dags:
            return
        log.info("Sync %s DAGs", len(dags))
        dag_by_ids = {dag.dag_id: dag for dag in dags}
        dag_ids = set(dag_by_ids.keys())
        query = (
            session.query(DagModel)
            .options(joinedload(DagModel.tags, innerjoin=False))
            .filter(DagModel.dag_id.in_(dag_ids))
            .options(joinedload(DagModel.schedule_dataset_references))
            .options(joinedload(DagModel.task_outlet_dataset_references))
        )
        orm_dags: list[DagModel] = with_row_locks(query, of=DagModel, session=session).all()
        existing_dags = {orm_dag.dag_id: orm_dag for orm_dag in orm_dags}