dag = dagbag.dags[request.dag_id]
        dag_run = dag.get_dagrun(run_id=request.run_id, session=session)
        dag.handle_callback(
            dagrun=dag_run, success=not request.is_failure_callback, reason=request.msg, session=session
        )
    def _execute_task_callbacks(self, dagbag: DagBag | None, request: TaskCallbackRequest, session: Session):
        if not request.is_failure_callback:
            return
        simple_ti = request.simple_task_instance
        ti: TI | None = (
            session.query(TI)
            .filter_by(
                dag_id=simple_ti.dag_id,
                run_id=simple_ti.run_id,
                task_id=simple_ti.task_id,
                map_index=simple_ti.map_index,
            )
            .one_or_none()
        )
        if not ti:
            return
        task: Operator | None = None
        if dagbag and simple_ti.dag_id in dagbag.dags:
            dag = dagbag.dags[simple_ti.dag_id]
            if simple_ti.task_id in dag.task_ids:
                task = dag.get_task(simple_ti.task_id)
        else: