self, callback_requests: list[CallbackRequest], session: Session
    ) -> None:
        """
        Execute what callbacks we can as "best effort" when the dag cannot be found/had parse errors.
        This is so important so that tasks that failed when there is a parse
        error don't get stuck in queued state.
        """
        for request in callback_requests:
            self.log.debug("Processing Callback Request: %s", request)
            if isinstance(request, TaskCallbackRequest):
                self._execute_task_callbacks(None, request, session)
            else:
                self.log.info(
                    "Not executing %s callback for file %s as there was a dag parse error",
                    request.__class__.__name__,
                    request.full_filepath,
                )
    @provide_session
    def _execute_dag_callbacks(self, dagbag: DagBag, request: DagCallbackRequest, session: Session):