This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
task = model.dag.get_task(simple_ti.task_id)
            except (exc.NoResultFound, TaskNotFound):
                pass
        if task:
            ti.refresh_from_task(task)
        ti.handle_failure(error=request.msg, test_mode=self.UNIT_TEST_MODE, session=session)
        self.log.info("Executed failure callback for %s in state %s", ti, ti.state)
        session.flush()
    @classmethod
    def _get_dagbag(cls, file_path: str):
        try:
            return DagBag(file_path, include_examples=False)
        except Exception:
            cls.logger().exception("Failed at reloading the DAG file %s", file_path)
            Stats.incr("dag_file_refresh_error", 1, 1)
            raise
    @provide_session
    def process_file(
        self,
        file_path: str,
        callback_requests: list[CallbackRequest],
        pickle_dags: bool = False,
        session: Session = NEW_SESSION,
    ) -> tuple[int, int]:
        """
        Process a Python file containing Airflow DAGs.
        This includes:
        1. Execute the file and look for DAG objects in the namespace.
        2. Execute any Callbacks if passed to this method.
        3. Serialize the DAGs and save it to DB (or update existing record in the DB).